Skip to content

Refactor FinGuard Governance and Router Pattern #21

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
google-labs-jules wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

Refactor FinGuard Governance and Router Pattern #21

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
61 changes: 2 additions & 59 deletions financial-advisor/financial_advisor/agent.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,71 +5,14 @@
from typing import AsyncIterator
import opentelemetry.trace as trace

from google.adk.agents import LlmAgent, InvocationContext
from google.adk.events.event import Event
from google.adk.tools.agent_tool import AgentTool
from pydantic import PrivateAttr
# Import from the Shared Governance Core
from vacp.governed_agent import VACPGovernedAgent

# Configure Logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
tracer = trace.get_tracer(__name__)

class VACPGovernedAgent(LlmAgent):
"""
An agent that delegates control decisions to the Verifiable Agentic Control Plane via OpenTelemetry.
"""

def __init__(self, **kwargs):
# Filter out extra args if needed, but LlmAgent accepts kwargs
super().__init__(**kwargs)

async def _run_async_impl(self, ctx: InvocationContext) -> AsyncIterator[Event]:
"""
Intercepts the agent's run loop to enforce VACP governance via OTel.
"""
logger.info(f"Agent {self.name} starting run under OTel-driven VACP governance.")

# Start Root Span for the Interaction
with tracer.start_as_current_span(f"agent.interaction.{ctx.invocation_id}") as root_span:
root_span.set_attribute("vacp.agent.id", self.name)
root_span.set_attribute("vacp.risk.tier", "High") # Dynamic in prod

# Helper to buffer reasoning text
reasoning_buffer = ""
reasoning_span = None

async for event in super()._run_async_impl(ctx):
# 1. Capture Reasoning (Thought Phase)
if event.content and event.content.parts:
text_content = event.content.parts[0].text
if text_content:
reasoning_buffer += text_content
# If we haven't started a reasoning span, start one
if not reasoning_span:
reasoning_span = tracer.start_span("gen_ai.reasoning")
reasoning_span.set_attribute("gen_ai.span.type", "reasoning")

# 2. Check for Tool Use (Action Phase)
# Heuristic: If event triggers a tool (simplified for ADK structure)
# In ADK, the event itself describes what happened.
# If we detect a tool call, we close the reasoning span.

# Note: This is a simplified OTel instrumentation for demonstration.
if hasattr(event, "tool_use") and event.tool_use:
if reasoning_span:
reasoning_span.set_attribute("gen_ai.content.completion", reasoning_buffer)
reasoning_span.end()
reasoning_span = None
reasoning_buffer = ""

yield event

# Clean up trailing reasoning
if reasoning_span:
reasoning_span.set_attribute("gen_ai.content.completion", reasoning_buffer)
reasoning_span.end()

# --- Initialization ---
try:
from . import prompt
Expand Down
33 changes: 19 additions & 14 deletions finguard/agents/compliance.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,16 +1,21 @@
from google.adk.agents import Agent
from finguard.tools.compliance import ComplianceTool
# Copyright 2025 Google LLC
# FinGuard Compliance Agent (Refactored for VACP)

def create_compliance_agent(model_client=None):
"""
Creates the Compliance Agent (Internal Auditor).
"""
compliance_tool = ComplianceTool()
from vacp.governed_agent import VACPGovernedAgent
from finguard.tools.compliance import validate_proposed_trade

compliance_agent = VACPGovernedAgent(
name="compliance_agent",
model="gemini-1.5-pro",
instruction="""
You are a Risk Officer and Compliance Auditor.
You do not generate code.
You only validate JSON payloads against the OPA policy.
Use the `validate_proposed_trade` tool to check if a trade is allowed.
""",
tools=[validate_proposed_trade],
description="Compliance Agent: Validates trades against risk policy."
)

return Agent(
name="compliance_agent",
model="gemini-1.5-pro",
instruction="You are a risk officer. You do not generate code. You only validate JSON payloads against the OPA policy. Use the 'validate_proposed_trade' tool.",
tools=[compliance_tool.validate_proposed_trade],
description="Compliance Agent: Validates trades against risk policy."
)
def create_compliance_agent(model_client=None):
return compliance_agent
209 changes: 63 additions & 146 deletions finguard/agents/coordinator.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,149 +1,66 @@
import asyncio
from typing import List, Dict, Any, Optional
from google.adk.agents import Agent

from finguard.governance.semantic_guard import SemanticGuard
from finguard.agents.compliance import create_compliance_agent
from finguard.agents.executor import create_executor_agent
from finguard.agents.quant import create_quant_agent
from finguard.agents.researcher import create_researcher_agent

# Copyright 2025 Google LLC
# FinGuard Coordinator (Refactored to match Financial Advisor Architecture)

import logging
from vacp.governed_agent import VACPGovernedAgent
from finguard.tools.router import route_request

# Import Sub-Agents (Workers)
from finguard.agents.researcher import researcher_agent
from finguard.agents.quant import quant_agent
from finguard.agents.compliance import compliance_agent
from finguard.agents.executor import executor_agent

logger = logging.getLogger(__name__)

# The Root Coordinator
# This replaces the manual 'FinGuardCoordinator' loop.
finguard_coordinator = VACPGovernedAgent(
name="finguard_coordinator",
model="gemini-1.5-pro", # Using 1.5 Pro as standard for FinGuard
description="FinGuard Supervisor: Manages risk and delegates tasks.",
instruction="""
You are the FinGuard Supervisor (Governance Layer).
Do NOT answer questions directly. Do NOT execute trades directly.

Your role is to route the session to the correct specialist based on the user's intent:
1. RESEARCH: If the user needs market news or data.
2. QUANT_ANALYSIS: If the user needs numerical analysis or Python code.
3. COMPLIANCE_CHECK: If the user proposes a trade or needs a policy check.
4. EXECUTION: If a trade is fully approved and ready to be placed.

Call the `route_request` tool with the correct intent and rationale.
""",
tools=[route_request],
sub_agents=[
researcher_agent,
quant_agent,
compliance_agent,
executor_agent
]
)

# For backward compatibility with existing tests/demos that might look for a class
class FinGuardCoordinator:
"""
The Supervisor Agent (Cortex).
Manages the lifecycle of a request, delegates to workers, and enforces semantic guardrails.
"""

def __init__(self, model_client: Any, project_id: Optional[str] = None):
self.model_client = model_client
self.semantic_guard = SemanticGuard(project_id=project_id, mock_mode=(project_id is None))

# Initialize Workers
self.compliance = create_compliance_agent(model_client)
self.executor = create_executor_agent(model_client)
self.quant = create_quant_agent(model_client)
self.researcher = create_researcher_agent(model_client)

# Inject Mock Client if provided (for testing in sandbox)
if model_client:
self._inject_client(self.compliance, model_client)
self._inject_client(self.executor, model_client)
self._inject_client(self.quant, model_client)
self._inject_client(self.researcher, model_client)

self.max_steps = 10
self.history = []

def _inject_client(self, agent: Agent, client: Any):
try:
agent._model_client = client
except Exception:
pass
try:
object.__setattr__(agent, 'client', client)
except Exception:
pass

async def run(self, user_query: str) -> str:
"""
Main execution loop.
"""
print(f"\n[COORDINATOR] Received: {user_query}")
self.history.append({"role": "user", "content": user_query})

step = 0
while step < self.max_steps:
step += 1

prompt = self._build_system_prompt()

# Call LLM (Simulated or Real)
try:
messages = [{"role": "system", "content": prompt}] + self.history[-5:]
# Check if chat is async
if asyncio.iscoroutinefunction(self.model_client.chat):
response_text = await self.model_client.chat(messages)
else:
response_text = self.model_client.chat(messages)
except Exception as e:
print(f"[COORDINATOR] LLM Error or Mock: {e}")
response_text = "I need to research the market first. Call Researcher."

print(f"[COORDINATOR] Thought: {response_text}")
Legacy Wrapper to maintain interface if needed, or we just replace usage.
For this refactor, we provide a way to 'run' which delegates to the agent runner?

# 2. Semantic Guard (Vaporwork Check)
drift_result = self.semantic_guard.check_drift(response_text)
if drift_result.is_drift:
print(f"[GOVERNANCE] HALT: Vaporwork detected. {drift_result.message}")
return f"Terminated due to repetitive behavior (Vaporwork). {drift_result.message}"

self.history.append({"role": "assistant", "content": response_text})

# 3. Parse & Route
if "Call Researcher" in response_text:
await self._delegate("Researcher", self.researcher, user_query)
elif "Call Quant" in response_text:
await self._delegate("Quant", self.quant, "Analyze AAPL volatility")
elif "Call Compliance" in response_text:
# For Policy Block test, we want to inject bad data if needed.
# But here we follow the prompt "Validate buy AAPL 1000".
await self._delegate("Compliance", self.compliance, "Validate buy AAPL 1000")
elif "Call Executor" in response_text:
await self._delegate("Executor", self.executor, "Execute buy AAPL 1000")
elif "Final Answer" in response_text:
return response_text
else:
pass

if step > 5:
break

return "Task completed (max steps reached)."

async def _delegate(self, name: str, agent: Agent, input_text: str):
print(f"[GOVERNANCE] Routing to {name}...")
try:
# ADK run_async expects keyword arguments matching inputs
# e.g. agent.run_async(prompt=input_text) or similar?
# InvocationContext usually takes a dict.
# LlmAgent.run_async(**inputs).
# If the prompt template expects 'input', we pass input=...
# If standard, maybe just pass dict?
# Let's try passing as user_message or similar.
# Usually input_text is treated as user message.

output_buffer = ""
# Assuming agent.run_async accepts kwargs that map to input.
# Or use 'prompt' or 'input'.
# LlmAgent usually handles user_content via InvocationContext if not provided?
# Let's try passing `input=input_text`.

# Note: run_async is an async generator
async for event in agent.run_async(prompt=input_text):
# Capture text content
if event.content and event.content.parts:
for part in event.content.parts:
if part.text:
output_buffer += part.text
# Also capture tool outputs if any (mocked)

if not output_buffer:
output_buffer = "[No text output]"

print(f"[{name}] Output: {output_buffer}")
self.history.append({"role": "system", "content": f"Tool {name} output: {output_buffer}"})
except Exception as e:
import traceback
# traceback.print_exc()
print(f"[{name}] Failed: {e}")

def _build_system_prompt(self) -> str:
return """You are the FinGuard Supervisor.
Plan the task.
If you need information, Call Researcher.
If you have data, Call Quant.
If you have a plan, Call Compliance.
If Compliance approves, Call Executor.

Respond with your thought and the next step.
"""
Actually, the 'main.py' or 'demo.py' usually instantiates this class.
We should update the consumer.
But to answer the user's requirement of "finguard becomes equivalent",
we expose the 'root_agent' (finguard_coordinator) as the entry point.
"""
def __init__(self, model_client=None, project_id=None):
self.agent = finguard_coordinator
# If model_client is passed (mocks), we might need to inject it?
# ADK agents usually get their client from the runtime configuration or environment.
pass

async def run(self, user_query: str):
# This emulates the old .run() method but uses the new ADK structure?
# ADK agents don't have a simple .run(str) -> str method like the old loop.
# They need a Runner.
# We should encourage using 'run_agent.py' or similar.
# But for drop-in replacement:
raise NotImplementedError("Use the ADK Runner to execute this agent.")
35 changes: 19 additions & 16 deletions finguard/agents/executor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,18 +1,21 @@
from google.adk.agents import Agent
from finguard.tools.execution import BrokerageTool
# Copyright 2025 Google LLC
# FinGuard Executor Agent (Refactored for VACP)

def create_executor_agent(model_client=None):
"""
Creates the Executor Agent (The Arm).
This is the ONLY agent with 'authorized=True' for the BrokerageTool.
"""
# ZSP: Injecting authorization here.
brokerage = BrokerageTool(authorized=True)
from vacp.governed_agent import VACPGovernedAgent
from finguard.tools.execution import execute_order, get_portfolio

executor_agent = VACPGovernedAgent(
name="executor_agent",
model="gemini-1.5-pro",
instruction="""
You are the Trade Executor.
You execute approved trades using the `execute_order` tool.
You must NOT execute if the trade has not been approved by Compliance.
You can check holdings with `get_portfolio`.
""",
tools=[execute_order, get_portfolio],
description="Executor Agent: Executes trades on the brokerage."
)

return Agent(
name="executor_agent",
model="gemini-1.5-pro",
instruction="You are the Trade Executor. You execute approved trades. You must NOT execute if the trade has not been approved by Compliance.",
tools=[brokerage.execute_order, brokerage.get_portfolio],
description="Executor Agent: Executes trades on the brokerage."
)
def create_executor_agent(model_client=None):
return executor_agent
34 changes: 19 additions & 15 deletions finguard/agents/quant.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,17 +1,21 @@
from google.adk.agents import Agent
from finguard.tools.quant import PythonSandboxTool
# Copyright 2025 Google LLC
# FinGuard Quant Agent (Refactored for VACP)

def create_quant_agent(model_client=None):
"""
Creates the Quant Agent (Analyst).
Running in a simulated sandbox.
"""
sandbox = PythonSandboxTool()
from vacp.governed_agent import VACPGovernedAgent
from finguard.tools.quant import run_python_analysis

quant_agent = VACPGovernedAgent(
name="quant_agent",
model="gemini-1.5-pro",
instruction="""
You are a Quantitative Analyst.
You write Python code to analyze financial data.
You do not have internet access.
Use the `run_python_analysis` tool for calculations.
""",
tools=[run_python_analysis],
description="Quant Agent: Performs numerical analysis using Python."
)

return Agent(
name="quant_agent",
model="gemini-1.5-pro",
instruction="You are a Quantitative Analyst. You write Python code to analyze financial data. You do not have internet access.",
tools=[sandbox.run_python_analysis],
description="Quant Agent: Performs numerical analysis using Python."
)
def create_quant_agent(model_client=None):
return quant_agent
Loading