Google ADK 深度探索（三）：Agent 架构——从单一职责到多智能体协作

# ✅ ADK 方式：一切都是 Python 对象
from google.adk.agents import Agent
import models

reviewer = Agent(
    model=models.moonshot,
    name="code_reviewer",
    instruction="你是一个严格的代码审查者",
    tools=[run_linter, check_security]
)

# ❌ 传统方式：配置文件 + 运行时解析
# agent_config.yaml:
# name: reviewer
# instruction: "你是一个严格的代码审查者"
# ...

# 团队负责人 Agent
team_lead = Agent(
    model=models.moonshot,
    name="lead",
    instruction="协调代码审查流程",
    sub_agents=[
        quality_checker,   # 质量检查
        security_auditor,  # 安全审计
        performance_tester # 性能测试
    ]
)

from google.adk.agents import Agent
import models

code_reviewer = Agent(
    model=models.moonshot,             # 大脑：用哪个模型推理
    name="code_reviewer",              # 身份：我是谁
    instruction="审查代码质量和安全性",  # 使命：我该做什么
    description="负责代码审查",         # 职责：我能做什么（给其他 Agent 看）
    tools=[run_linter],                # 能力：我有哪些工具
    sub_agents=[security_agent]        # 团队：我管理哪些下属
)

# 使用 pip
pip install google-adk

# 或使用 uv（推荐）
uv pip install google-adk

mkdir my_agent && cd my_agent
touch agent.py .env __init__.py

from google.adk.agents import Agent
from google.adk.models import LiteLlm

moonshot = LiteLlm(model="moonshot/moonshot-v1-128k")

root_agent = Agent(
    model=moonshot,
    name="greeter",
    description="A friendly assistant",
    instruction="你是一个友好的助手，请简洁地回答问题",
)

# .env
MOONSHOT_API_KEY=sk-xxxx
MOONSHOT_API_BASE=https://api.moonshot.cn/v1

# 或 DeepSeek
DEEPSEEK_API_KEY=sk-xxxx

# 从父目录运行
cd ..
adk run my_agent

adk web --port 8888 --reload_agents

from google.adk.agents import Agent
from google.adk.models import LiteLlm

moonshot = LiteLlm(model="moonshot/moonshot-v1-128k")

# 工具 1：代码风格检查
def check_code_style(code: str, language: str) -> dict:
    """检查代码风格是否符合规范"""
    issues = []
    if language == "python":
        if "import *" in code:
            issues.append("避免使用 'import *'，应明确导入")
        if not code.strip().endswith("\n"):
            issues.append("文件应以换行符结尾")
    
    return {
        "status": "success" if not issues else "warning",
        "issues": issues,
        "checked_lines": len(code.split("\n"))
    }

# 工具 2：安全漏洞扫描
def scan_security(code: str) -> dict:
    """扫描常见安全问题"""
    vulnerabilities = []
    if "eval(" in code:
        vulnerabilities.append("发现 eval() 调用，存在代码注入风险")
    if "password" in code.lower() and "=" in code:
        vulnerabilities.append("检测到硬编码密码，建议使用环境变量")
    
    return {
        "status": "critical" if vulnerabilities else "safe",
        "vulnerabilities": vulnerabilities
    }

root_agent = Agent(
    model=moonshot,
    name="code_reviewer",
    description="专业代码审查助手，检查代码风格和安全问题",
    instruction="""你是一个严格但友好的代码审查者。

职责：
1. 使用 check_code_style 工具检查代码风格
2. 使用 scan_security 工具扫描安全漏洞
3. 综合两个工具的结果，给出清晰的审查报告

报告格式：
- 代码风格：[问题列表或"无问题"]
- 安全检查：[漏洞列表或"无漏洞"]
- 建议：[具体改进建议]
""",
    tools=[check_code_style, scan_security]
)

def check_code_style(code: str, language: str) -> dict:
    """检查代码风格是否符合规范"""

def scan_security(code: str) -> dict:
    """扫描常见安全问题"""  # ← LLM 看到这个描述决定是否调用

# ✅ 推荐
return {"status": "warning", "issues": ["问题1", "问题2"]}

# ❌ 不推荐
return "发现 2 个问题：问题1，问题2"

adk run code_reviewer

def check_code_style(code: str, language: str) -> dict:
    """检查代码风格是否符合规范"""
    # 你的业务逻辑
    return {"issues": [...], "count": 10}

from google.adk.tools.openapi_tool import OpenAPIToolset

# 示意代码，spec_str 替换为实际 OpenAPI 规范内容
github_tools = OpenAPIToolset(
    spec_str='{"openapi": "3.0.0", "...": "..."}',
    spec_str_type="json"
)

agent = Agent(
    model=moonshot,
    name="github_bot",
    tools=github_tools.get_tools()
)

from google.adk.tools.mcp_tool import McpToolset
from google.adk.tools.mcp_tool.mcp_toolset import StreamableHTTPConnectionParams

mcp_tools = McpToolset(
    connection_params=StreamableHTTPConnectionParams(
        url="https://your-mcp-server-url/mcp",
        headers={"Authorization": "Bearer your-auth-token"}
    )
)

agent = Agent(
    model=moonshot,
    name="agent_with_mcp",
    tools=mcp_tools.get_tools()
)

from google.adk.agents import Agent
from google.adk.models import LiteLlm

moonshot = LiteLlm(model="moonshot/moonshot-v1-128k")

# 子 Agent 1：代码质量检查
# description 要具体，父 Agent 靠它决定何时委派
quality_checker = Agent(
    model=moonshot,
    name="quality_checker",
    description="检查代码质量和风格规范",  # ← 清晰的职责边界
    instruction="分析代码可读性、命名规范、注释完整性",
    tools=[check_code_style]
)

# 子 Agent 2：安全审计
security_auditor = Agent(
    model=moonshot,
    name="security_auditor",
    description="扫描 SQL 注入、XSS、硬编码密码等安全漏洞",  # ← 具体到漏洞类型
    instruction="识别常见安全问题并给出修复建议",
    tools=[scan_security]
)

# 父 Agent：协调者
code_reviewer = Agent(
    model=moonshot,
    name="code_reviewer",
    description="代码审查协调者，统筹质量检查和安全审计流程",
    instruction="""你是代码审查的协调者。

工作流程：
1. 将代码质量检查任务委派给 quality_checker
2. 将安全审计任务委派给 security_auditor
3. 综合两者报告，生成最终审查结果

注意：如发现严重安全问题，优先级高于代码风格问题。
""",
    sub_agents=[quality_checker, security_auditor]
)

from google.adk.agents.sequential_agent import SequentialAgent

step1_analyzer = Agent(
    model=moonshot,
    name="analyzer",
    description="分析需求文档，提取关键需求和约束",
    instruction="提取核心需求和约束条件"
)

step2_designer = Agent(
    model=moonshot,
    name="designer",
    description="根据需求设计技术方案和接口",
    instruction="设计架构和接口定义"
)

step3_reviewer = Agent(
    model=moonshot,
    name="reviewer",
    description="评审方案可行性，识别技术风险",
    instruction="评估技术风险和实现难度"
)

workflow = SequentialAgent(
    name="requirement_workflow",
    description="需求分析 → 方案设计 → 可行性评审",
    sub_agents=[step1_analyzer, step2_designer, step3_reviewer]
)

from google.adk.agents.parallel_agent import ParallelAgent

ci_pipeline = ParallelAgent(
    name="ci_checks",
    description="CI 流水线：代码检查 + 类型检查 + 测试同时运行",
    sub_agents=[lint_checker, type_checker, test_runner]
)

# ✅ 好
quality_checker = Agent(description="检查代码质量")
security_auditor = Agent(description="扫描安全漏洞")

# ❌ 不好
super_agent = Agent(description="检查质量、安全、性能、文档...")

# ✅ 好
Agent(description="扫描 SQL 注入、XSS、硬编码密码等安全漏洞")

# ❌ 不好
Agent(description="处理安全相关的事情")

# 通过 SessionService 创建 Session
session_service = InMemorySessionService()
session = await session_service.create_session(
    app_name="my_app", user_id="u1"
)

session.state["progress"] = "50%"          # 无前缀：当前会话
session.state["user:theme"] = "dark"       # user: 跨会话保留
session.state["app:version"] = "1.0"       # app: 全局配置
session.state["temp:file_content"] = "..."  # temp: 仅当前调用

from google.adk.tools.tool_context import ToolContext

def analyze_with_context(tool_context: ToolContext) -> str:
    """读取 Context 状态"""
    # 通过 tool_context.state 直接访问 session state
    lang = tool_context.state.get("user:language_preference", "en")
    
    # 读取会话状态
    files = tool_context.state.get("files_uploaded", [])
    
    return f"已处理 {len(files)} 个文件，语言：{lang}"

# 父 Agent 写入
parent_agent = Agent(
    instruction="""协调代码审查流程。
将上传的文件列表保存到 session.state['files']，
子 Agent 会自动读取并处理。
""",
    sub_agents=[quality_checker, security_auditor]
)

# 子 Agent 读取（无需显式传参）
quality_checker = Agent(
    instruction="从 session.state['files'] 读取文件列表并检查质量"
)

from google.adk.tools.load_artifacts_tool import load_artifacts_tool

file_analyzer = Agent(
    model=moonshot,
    name="analyzer",
    instruction="使用 load_artifacts_tool 加载用户上传的文件并分析",
    tools=[load_artifacts_tool]
)

# ✅ 好
session.state["user:theme"] = "dark"   # 明确是用户偏好

# ❌ 不好
session.state["theme"] = "dark"        # 会话结束就丢失了

# ✅ 好
session.state["temp:intermediate"] = {...}  # 调用结束自动清理

# ❌ 不好
session.state["intermediate"] = {...}       # 污染会话状态

tool_context.state["temp:step_progress"] = "步骤 1：加载文件完成，发现 3 个问题"
# temp: 前缀确保仅在当前调用中可见，不会污染会话状态
# 在 adk web 中可以通过 session state 查看调试信息

# 传统方式：显式控制每一步
if "quality" in request:
    check_quality()
if "security" in request:
    check_security()
combine_results()

# ADK 方式：声明目标，让 Agent 推理路径
Agent(
    instruction="审查代码质量和安全性",
    sub_agents=[quality_checker, security_auditor]
)