Agent Runtime & Frameworks 调研

from google.adk.agents import LlmAgent

# 1. 定义感知工具：赋予 Agent 探索世界的能力
def get_capital_city(country: str) -> str:
    """查询指定国家的首都。

    Args:
        country: 国家名称（支持中文和英文）
    """
    capitals = {"france": "Paris", "中国": "北京", "japan": "Tokyo"}
    return capitals.get(country, f"数据库中未找到: {country}")

# 2. 构建认知模型：注入角色设定与思维链
geographer = LlmAgent(
    model="gemini-2.5-flash",  # 选择推理基座
    name="geographer_agent",   # 设定唯一身份标识
    description="专注于地理知识的查询与解答",

    # System Prompt 即 Agent 的"灵魂"
    instruction="""
    你是一个严谨的地理学家。
    1. 在回答问题前，先判断是否需要使用工具。
    2. 如果查询结果不存在，请诚实地告知用户，不要编造。
    3. 对于工具返回的数据，结合你的知识库进行补充说明（如人口、气候等）。
    """,

    tools=[get_capital_city]   # 挂载认知工具
)

from google.adk.agents import BaseAgent
from google.adk.agents.invocation_context import InvocationContext
from google.adk.events import Event
from typing import AsyncGenerator

class ComplianceAgent(BaseAgent):
    """一个不使用 LLM，但拥有『一票否决权』的规则智能体"""

    strictly_mode: bool = True

    async def _run_async_impl(self, context: InvocationContext) -> AsyncGenerator[Event, None]:
        # 1. 获取上下文中的"记忆"
        draft_content = context.session.state.get("draft_report", "")

        # 2. 执行确定性的逻辑 (Symbolic Logic)
        sensitives = self._scan_sensitive_words(draft_content)

        if sensitives:
            # 拒绝通过，并返回结构化的反馈
            yield Event(
                author=self.name,
                content=f"REJECTED: Detect sensitive words: {sensitives}. Please sanitize the content."
            )
            return

        # 3. 更新状态，允许流程继续
        context.session.state["compliance_passed"] = True
        yield Event(author=self.name, content="APPROVED")

    def _scan_sensitive_words(self, text: str) -> list:
        # 实现高效的 AC 自动机或正则匹配
        return [w for w in ["机密", "Top Secret"] if w in text]

from google.adk.agents import LlmAgent
from google.adk.tools import FunctionTool

# ✅ 优雅定义：通过 Type Hints 和 Docstring 自动生成 Schema
def lookup_sales_data(region: str, quarter: str = "Q1") -> dict:
    """查询指定区域的季度销售数据。

    用于在回答财务相关问题时，获取准确的业务报表数据。

    Args:
        region: 目标区域代码 (如 'CN-NORTH', 'US-WEST')
        quarter: 财务季度 (如 '2025-Q1')

    Returns:
        包含 'revenue', 'cost', 'profit' 的字典
    """
    # 实际业务逻辑连接到 ERP 系统
    return erp_client.query(region, quarter)

# ADK 自动处理转换
agent = LlmAgent(
    name="financial_assistant",
    tools=[lookup_sales_data]  # 直接挂载函数，ADK 自动包装为 FunctionTool
)

from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset
from mcp import StdioServerParameters

# 一键接入：连接现有的 MCP Server（需在 async 上下文中调用）
tools, exit_stack = await MCPToolset.from_server(
    connection_params=StdioServerParameters(
        command="npx",
        args=["-y", "@modelcontextprotocol/server-github"]
    )
)

dev_agent = LlmAgent(
    model="gemini-2.5-flash",
    name="git_ops_agent",
    tools=tools  # 获得一系列 Git 操作工具
)

from google.adk.memory import VertexAiMemoryBankService

# 🧠 接驳云端海马体
hippocampus = VertexAiMemoryBankService(
    project="aurelius-agent-platform",
    location="us-central1",
    agent_engine_id="your-agent-engine-id"
)

# 注入具备长期记忆的 Agent
consultant = LlmAgent(
    model="gemini-2.5-flash",
    name="senior_consultant",
    memory_service=hippocampus,  # 自动具备 RAG 能力
    instruction="在回答问题前，先从记忆库中回忆相关的历史案例..."
)

import google.genai.types as types
from google.adk.tools.tool_context import ToolContext

# 注入 ToolContext，让工具具备"记忆"和"创造"能力
async def generate_analysis_report(data_id: str, ctx: ToolContext) -> str:
    """生成并保存数据分析报告。

    Args:
        data_id: 数据集 ID
        ctx: ADK 自动注入的上下文对象 (Magic Argument)
    """
    # 1. Access State: 获取用户的身份信息
    user_tier = ctx.state.get("user_tier", "standard")

    # 2. Side Effect: 生成 PDF 文件并作为 Artifact 保存
    report_bytes = _run_analysis(data_id, depth=user_tier)
    report_artifact = types.Part.from_bytes(
        data=report_bytes, mime_type="application/pdf"
    )
    version = await ctx.save_artifact(
        filename=f"report_{data_id}.pdf",
        artifact=report_artifact
    )

    # 3. Mutate State: 更新任务进度
    ctx.state["last_report_url"] = f"report_{data_id}.pdf"

    return f"报告已生成完毕，您可以点击此处下载: report_{data_id}.pdf"

from google.adk.agents import LlmAgent, SequentialAgent

# 1. 组建专家团队 (The Specialists)
# 每一位专家都拥有独特的"人设"和"技能树"

# 研究员：追求速度与广度，使用 Flash 模型
researcher = LlmAgent(
    name="info_gatherer",
    model="gemini-2.5-flash",
    instruction="你的目标是广度优先地收集信息，确保覆盖所有相关事实，不放过任何细节。",
    tools=[google_search_tool, vector_db_query_tool]
)

# 分析师：追求逻辑与深度，使用 Pro 模型
analyst = LlmAgent(
    name="insight_extractor",
    model="gemini-2.5-pro",  # 使用更强的推理模型
    instruction="""
    你需要基于研究员提供的事实，进行深度思考：
    1. 识别数据背后的模式与反直觉的结论。
    2. 剔除噪音，提炼核心洞察。
    """,
)

# 笔者：追求表达与风格，使用 Flash 模型
writer = LlmAgent(
    name="final_editor",
    model="gemini-2.5-flash",
    instruction="将枯燥的洞察转化为引人入胜的行业报告，保持客观、专业的语调。",
)

# 2. 定义协作协议 (The Protocol)
# 将专家串联为一条"深度研究流水线"
# 关键点：这个 Pipeline 本身也是一个 Agent，可以被上层架构继续集成（分形特性）
deep_research_squad = SequentialAgent(
    name="deep_research_pipeline",
    description="自动执行从信息搜集到报告生成的端到端任务",
    sub_agents=[researcher, analyst, writer]
)

# Server Side: 将本地 Agent 暴露为 A2A 服务
# ADK 通过 a2a.utils 模块提供 A2A 暴露能力
from google.adk.a2a.utils.agent_to_a2a import to_a2a

# 启动一个专职的"数据分析服务"
# to_a2a 封装了 A2A 协议所需的标准接口调用
a2a_server = to_a2a(agent=analyst_agent)

# Client Side: 像调用本地函数一样调用远程智能体
# 通过 a2a-sdk 独立包提供的 A2AClient 连接远程 Agent
from a2a.client import A2AClient

# 连接远程专家 (自动处理握手)
remote_analyst = A2AClient(
    url="https://analyst-agent.internal.corp"
)

# 远程调用：包含完整的上下文传递
analysis = await remote_analyst.send_task(
    message="分析这份财报的异常点",
    context={"report_url": "s3://..."}
)

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions

async def main():
    # 建立连接：开启一段自主任务
    stream = query(
        prompt="在 auth.py 中找到并修复 bug",
        # 飞行控制：定义 Agent 的权限边界
        options=ClaudeAgentOptions(
            allowed_tools=["Read", "Edit", "Bash"], # 赋予系统权限
            permission_mode="acceptEdits"           # 授予自动修改权 (Autonomy)
        )
    )

    # 实时观察：监听 Agent 的思考脉搏
    async for event in stream:
        # event 可能是：思考片段 (Thought)、工具执行 (Action)、或者最终结果 (Result)
        print(f"[{event.type}] {event.content}")

        # 你甚至可以在这里进行"在路干预" (Human-in-the-loop)
        # if event.type == "UserPermissionRequired": ...

asyncio.run(main())

import { query } from "@anthropic-ai/claude-agent-sdk";

async function main() {
  const stream = query({
    prompt: "Find and fix the bug in auth.py",
    options: {
      allowedTools: ["Read", "Edit", "Bash"],
      permissionMode: "acceptEdits",
    },
  });

  // 同样使用 for-await 语法处理流
  for await (const message of stream) {
    console.log(message);
  }
}

main();

# The Analyst (绝对安全)
# 只能看，不能动。适合做代码审查、文档生成或架构分析。
profile_analyst = ClaudeAgentOptions(
    allowed_tools=["Glob", "Grep", "Read"],  # 剥夺修改与执行权
    permission_mode="default"                 # 读操作无需审批
)

# The Refactorer (代码风格)
# 可以改代码，但不能运行命令。适合做重构、格式化或添加注释，风险可控。
profile_refactorer = ClaudeAgentOptions(
    allowed_tools=["Glob", "Grep", "Read", "Edit"],
    permission_mode="acceptEdits"             # 授权自动修改文件
)

# The Full Stack Engineer (上帝模式)
# 拥有完整权限。适合做端到端开发、测试驱动开发 (TDD) 或环境配置。
profile_engineer = ClaudeAgentOptions(
    allowed_tools=["Glob", "Grep", "Read", "Edit", "Bash"],
    permission_mode="bypassPermissions"       # 完全授权（仅限沙箱环境）
)

from claude_agent_sdk import PermissionResultAllow, PermissionResultDeny

async def smart_firewall(tool_name: str, args: dict, context) -> PermissionResultAllow | PermissionResultDeny:
    """语义级安全拦截器

    不仅仅是正则匹配，而是理解 Agent 的意图。
    """
    if tool_name == "Bash":
        command = args.get("command", "")

        # 1. 拦截高危操作 (即使是 root 用户也无法绕过)
        if any(risk in command for risk in ["rm -rf /", "mkfs", ":(){ :|:& };:"]):
            print(f"拦截到自毁指令: {command}")
            return PermissionResultDeny()

        # 2. 限制网络外联 (防止数据泄露)
        if "curl" in command or "wget" in command:
            # 仅允许访问内部受信任的域名
            if "internal.corp" not in command:
                print(f"拦截到外部网络请求: {command}")
                return PermissionResultDeny()

    return PermissionResultAllow()

# 注入防火墙
options = ClaudeAgentOptions(
    permission_mode="default",       # 基础模式
    can_use_tool=smart_firewall      # 叠加自定义拦截逻辑
)

my-project/
├── .claude/
│   ├── skills/               # 💪 肌肉记忆 (Skillset)
│   │   └── deploy_gke.md     # 定义复杂的行动SOP，如"如何部署到GKE"
│   └── commands/             # ⚡ 条件反射 (shortcuts)
│       └── fix-lint.md       # 定义快捷指令，如 "/lint"
├── CLAUDE.md                 # 🧠 长期记忆 (Project Memory)
│                             # 存放项目架构规范、编码风格、常用术语
└── ...

# 为 Agent 注入 SQL 读写能力
options = ClaudeAgentOptions(
    mcp_servers={
        "corporate_memory": {
            "command": "uvx",  # 使用高性能包管理器
            "args": [
                "mcp-server-postgres",
                "--connection-string",
                "postgresql://user:secret@localhost/db"
            ]
        }
    }
)

## <!-- L1: 放在文件头部的元数据，Agent 随时都能"看到"这部分 -->

name: pdf-processing
description: 从 PDF 文件中提取文本和表格，填写表单。在用户提到发票、合同或扫描件时激活。

---

<!-- L2: 具体的技能正文，平时是折叠的，只有被激活后才会展开进入 Context -->

# PDF Processing Guide

当处理 PDF 文档时，请遵循以下流程：

1. 首先使用 `pdftotext` 尝试提取文本。
2. 如果文本乱码，则调用 `ocr.py` 脚本（L3 资源）。
   ...

.claude/skills/pdf/
├── SKILL.md           # L2: 主指令入口
├── scripts/           # L3: 工具箱 (按需执行)
│   └── ocr.py
└── docs/              # L3: 参考资料 (按需查阅)
    └── invoice_codes.txt

# 场景：通过 Agent SDK，只需一句话自动化完成复杂的报表工作
# Skills 随 Agent 环境自动加载，无需显式配置 tools
async for msg in query(
    prompt="""
    请帮我把这个月的数据库导出文件（data.csv）做一次深度清洗：
    1. 修复所有格式错误的电话号码。
    2. 生成一个带数据透视表的 Excel 文件 (report.xlsx)。
    3. 用邮件把这个 Excel 发给 boss@company.com。
    """,
    options=ClaudeAgentOptions(
        allowed_tools=["Read", "Edit", "Bash"],
        permission_mode="acceptEdits"
    )
):
    pass  # Skills 随 Agent 环境内置，无需额外配置

---
# 🎯 触发器 (The Trigger)
# Frontmatter 定义了技能的"人设"和"激活条件"
name: senior-code-reviewer
description: 扮演一位严厉的资深架构师，对 Python/Go 代码进行安全性与性能审查。
  当用户提交 PR 或请求 audit 时激活。
---

# 🧠 执行逻辑 (The Logic)

# 正文定义了技能的"思考框架"和"输出标准"

## Review Philosophy

不要纠结于细枝末节的格式问题（交给 Linter），请专注于：

1. **Security**: 识别 SQL 注入、XSS、敏感信息泄露。
2. **Performance**: 识别 O(n^2) 复杂度、内存泄漏风险。
3. **Maintainability**: 识别过度设计、硬编码。

## Interaction Protocol

在审查前，先运行 `git diff --stat` 了解变更范围。
如果发现 critical issue，请直接通过 `Edit` 工具修复代码，而不仅仅是提出建议。

## Report Template

请严格遵循以下输出格式：

```markdown
## 🚨 Security Audit Report

> Severity Level: HIGH/MEDIUM/LOW

### ⛔ Blocking Issues

- [ ] file/path.py:L23 - SQL Injection detected

### ⚠️ Optimization

- file/path.py:L45 - List comprehension is faster here
```

# 🏗️ 初始化工程结构
mkdir agent-factory && cd agent-factory

# 📦 安装核心套件
# 建议使用 uv 或 poetry 进行依赖管理
python -m venv .venv
source .venv/bin/activate
pip install google-adk

# 🔑 颁发入厂证 (Google Cloud Auth)
# 确保你的账号具有 Vertex AI User 权限
gcloud auth application-default login

# factory/workers/researcher.py
from google.adk.agents import LlmAgent
from google.adk.tools import google_search

def hire_researcher() -> LlmAgent:
    """招聘一名初级研究员 (Standard Worker)"""
    return LlmAgent(
        model="gemini-2.5-flash",  # 速度优先：快速扫描大量信息
        name="research_worker_01",

        # 📜 SOP: 把模糊的指令转化为确定的动作
        instruction="""
        Role: Corporate Researcher (Level 1)

        Objective:
        收集原始数据，建立事实基础。严禁臆造，必须有据可查。

        Standard Operating Procedure (SOP):
        1. **Query Expansion**: 将用户问题拆解为 3-5 个正交的搜索关键词。
        2. **Cross-Examination**: 对同一个事实，寻找至少两个独立来源进行交叉验证。
        3. **Data Structuring**: 将非结构化网页内容转化为 Bullet Points。

        Output Constraint:
        - Format: JSON-like structure with {Fact, Source, Confidence}.
        - No opinion, just facts.
        """,
        tools=[google_search]
    )

def hire_analyst() -> LlmAgent:
    """招聘一名高级分析师 (Thinking Unit)"""
    return LlmAgent(
        model="gemini-2.5-pro",  # 推理优先：擅长逻辑链
        name="analyst_worker_01",
        instruction="""
        Role: Senior Data Analyst (Level 3)

        Objective:
        负责"Raw Data -> Insight"的价值转化。

        Reasoning Frameworks:
        - **Pattern Recognition**: 识别数据中的异常值和重复模式。
        - **Second-Order Thinking**: 不仅看现象，更要推演其长远影响。
        - **Conflict Identification**: 指出搜索结果中相互矛盾的信息点。

        Output Interface:
        输出一段包含 "Key Findings" 和 "Strategic Implications" 的深度分析。
        """,
        tools=[] # 纯脑力劳动，无需外设
    )

def hire_writer() -> LlmAgent:
    """招聘一名特稿撰写人 (Creative Unit)"""
    return LlmAgent(
        model="gemini-2.5-pro",  # 文笔优先：擅长修辞与结构
        name="lead_writer",
        instruction="""
        Role: Lead Tech Columnist

        Objective:
        将枯燥的分析转化为具有传播力的深度文章。

        Editorial Standards:
        1. **The Hook**: 开篇必须在 3 秒内抓住读者注意力。
        2. **Structure**: 使用金字塔原理 (Pyramid Principle) 组织段落。
        3. **Tone**: 专业但机智 (Professional yet witty)。避免陈词滥调 (Clichés)。
        4. **Formatting**: 熟练使用 Markdown (Bold, Quote, Table) 增强可读性。
        """,
        tools=[]
    )

def hire_designer() -> LlmAgent:
    """招聘一名数据可视化专家 (Visual Unit)"""
    return LlmAgent(
        model="gemini-2.5-flash",
        name="visual_designer",
        instruction="""
        Role: Information Designer

        Objective:
        将抽象逻辑转化为直观的视觉图表。

        Design System:
        - Tool: Mermaid.js
        - Allowed Types: `graph TB` (流程架构), `pie` (占比), `gantt` (时间线).
        - Style: 简洁现代，避免复杂的嵌套。

        Execution:
        仅输出代码块 (```mermaid ... ```)，严禁包含任何解释性文字。
        """,
        tools=[]
    )

# factory/pipelines/report_pipeline.py
from google.adk.agents import SequentialAgent, ParallelAgent
from .workers import hire_researcher, hire_analyst, hire_writer, hire_designer

def build_assembly_line():
    """生产一份研报的完整流水线"""

    # Stage 1: 原材料获取
    researcher = hire_researcher()

    # Stage 2: 粗加工 (提炼洞察)
    analyst = hire_analyst()

    # Stage 3: 精加工 (并行作业)
    # 文字与图表同时制作，提高吞吐量
    production_floor = ParallelAgent(
        name="creative_process",
        sub_agents=[
            hire_writer(),   # 写手
            hire_designer()  # 画师
        ]
    )

    # 最终组装
    return SequentialAgent(
        name="daily_report_pipeline",
        sub_agents=[researcher, analyst, production_floor]
    )

# main_factory.py
import asyncio
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from factory.pipelines.report_pipeline import build_assembly_line

async def start_production():
    # 启动流水线
    pipeline = build_assembly_line()
    session_service = InMemorySessionService()
    runner = Runner(agent=pipeline, session_service=session_service)

    print("Factory started...")

    # 创建会话
    session = await session_service.create_session(
        user_id="factory_user",
        app_name="report_factory"
    )

    # 投料
    from google.genai import types

    async for event in runner.run_async(
        user_id="factory_user",
        session_id=session.id,
        new_message=types.Content(
            role="user",
            parts=[types.Part.from_text(text="深度分析 DeepSeek-V3 对 Transformer 架构的改进")]
        )
    ):
        if event.is_final_response():
            print("Product shipped!")
            print(event.content)

if __name__ == "__main__":
    asyncio.run(start_production())

# 准备宿主环境
mkdir claude-symbiont && cd claude-symbiont

# 唤醒 Core (Claude Agent SDK 自动捆绑 CLI，无需单独安装)
# SDK 默认使用内置的 Claude Code CLI

# 注入 SDK
# 强烈推荐使用 uv，极其快速
uv init
uv add claude-agent-sdk

# symbiont.py
import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions

async def smart_review(target_dir: str):
    """
    一个拥有独立视角的代码审查者
    它可以"看到"文件，"理解"逻辑，并给出建议。
    """

    # 开启思维流
    stream = query(
        prompt=f"""
        Mission: Audit Python code in '{target_dir}' for security vulnerabilities.

        Focus Areas:
        1. SQL Injection (raw queries without parametrization)
        2. Hardcoded Secrets (API keys, passwords)
        3. Unsafe Deserialization (pickle.load)

        Output:
        Generate a Markdown report summarizing findings with severity levels.
        """,
        options=ClaudeAgentOptions(
            # 只给它看的权限，不给改的权限 (Read-Only)
            allowed_tools=["Read", "Glob", "Grep"],
            permission_mode="default" # 对于只读操作，SDK 默认会智能放行
        )
    )

    print(f"开始审计 {target_dir}...")

    async for event in stream:
        # 实时打印它的思考过程
        print(event)

if __name__ == "__main__":
    asyncio.run(smart_review("./src"))

# augmented_symbiont.py
import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions

async def office_automation():
    """
    一个掌握了 Office 技能的自动化助理
    """

    # 注入项目级配置 (包含 .claude/skills/*)
    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Edit", "Bash"], # 允许它写文件、运行命令
        permission_mode="acceptEdits"           # 授予自动执行权
    )

    # 自然语言指令
    prompt = """
    读取 data/sales_q4.csv，做以下处理：
    1. 用 Pandas 分析各地区的销售环比增长。
    2. 生成一个漂亮的 Matplotlib 柱状图。
    3. 最后生成一个总结性的 Excel 报表 (summary.xlsx)。
    """

    print("正在执行办公自动化任务...")
    async for msg in query(prompt=prompt, options=options):
        pass # 静默执行，只看结果

if __name__ == "__main__":
    asyncio.run(office_automation())

# 像创建 Dockerfile 一样创建 Skill
mkdir -p .claude/skills/data-science
touch .claude/skills/data-science/SKILL.md

# 定义"数据科学家"的人设与SOP
cat > .claude/skills/data-science/SKILL.md << 'EOF'
---
name: data-scientist
description: 专业的 Python 数据分析师，擅长 Pandas/Numpy/Matplotlib。
  当用户涉及 csv 处理、统计分析或绘图时激活。
---

# Data Science Protocol

## Workflow
1. **Inspection**: 先打印 DataFrame 的 `head()` 和 `info()`。
2. **Cleaning**: 检查并处理 NaN 值。
3. **Visualization**: 所有图表必须有 Title, Label 和 Legend。
4. **Output**: 尽量保存为 high-dpi 的 vector graphics (SVG/PDF)。

## Code Snippet
在使用 Pandas 时，请优先使用链式调用 (Method Chaining) 以保持代码整洁。
EOF