MultiAgent 协作模式全解析：从单体 Agent 到分布式 AI 系统

# Agent A 写入
session_state["analysis_result"] = "竞品A的核心优势在于..."

# Agent B 读取
result = session_state.get("analysis_result")

# Sub-agent 被包装为 Tool
class DataAnalysisAgentTool(Tool):
    def run(self, data: str) -> str:
        return data_analysis_agent.run(data)

# Orchestrator 像调用普通工具一样调用它
orchestrator.tools = [DataAnalysisAgentTool(), ...]

"""
示例：三步内容生成流水线
步骤1: 生成文章大纲
步骤2: 基于大纲生成正文
步骤3: 润色优化
"""

from google.adk.agents import SequentialAgent, LlmAgent
from google.adk.sessions import InMemorySessionService
from google.adk.runners import Runner
from google.genai import types

# 步骤1: 大纲生成 Agent
outline_agent = LlmAgent(
    name="OutlineAgent",
    model="gemini-2.0-flash",
    instruction="""你是一位专业的内容策划师。
    用户会给你一个主题，请生成一份结构清晰的文章大纲（3-5个主要章节）。
    输出格式：仅输出大纲内容，每个章节标题单独一行。""",
    output_key="article_outline"  # 输出写入 session state
)

# 步骤2: 正文生成 Agent
draft_agent = LlmAgent(
    name="DraftAgent",
    model="gemini-2.0-flash",
    instruction="""你是一位专业的文章撰写师。
    请基于 session state 中 'article_outline' 的大纲，撰写完整的文章正文。
    要求：每个章节至少200字，语言流畅自然。""",
    output_key="article_draft"
)

# 步骤3: 润色 Agent
polish_agent = LlmAgent(
    name="PolishAgent",
    model="gemini-2.0-flash",
    instruction="""你是一位资深文字编辑。
    请对 session state 中 'article_draft' 的文章进行润色：
    1. 改善句子流畅度
    2. 增强表达的专业性
    3. 确保段落衔接自然
    输出润色后的最终版本。""",
    output_key="final_article"
)

# 组合为 SequentialAgent
pipeline = SequentialAgent(
    name="ContentPipeline",
    sub_agents=[outline_agent, draft_agent, polish_agent]
)

# 运行
session_service = InMemorySessionService()
runner = Runner(agent=pipeline, session_service=session_service, app_name="content_app")

session = session_service.create_session(app_name="content_app", user_id="user_1")
response = runner.run(
    user_id="user_1",
    session_id=session.id,
    new_message=types.Content(parts=[types.Part(text="多智能体系统的未来发展")])
)

print("=== 生成结果 ===")
print(session_service.get_session(
    app_name="content_app", user_id="user_1", session_id=session.id
).state.get("final_article"))

"""
示例：三维度并行产品评估
同时从「用户体验」「技术可行性」「商业价值」三个维度评估产品方案
"""

from google.adk.agents import ParallelAgent, LlmAgent, SequentialAgent
from google.adk.sessions import InMemorySessionService
from google.adk.runners import Runner
from google.genai import types

# 三个并行的专家 Agent
ux_expert = LlmAgent(
    name="UXExpert",
    model="gemini-2.0-flash",
    instruction="""你是一位资深用户体验专家。
    请从用户体验角度评估以下产品方案（100字以内）：
    评估维度：易用性、学习曲线、用户满意度预期。
    输出格式：[UX评分: X/10] 评估内容""",
    output_key="ux_assessment"
)

tech_expert = LlmAgent(
    name="TechExpert",
    model="gemini-2.0-flash",
    instruction="""你是一位资深技术架构师。
    请从技术可行性角度评估以下产品方案（100字以内）：
    评估维度：实现难度、技术风险、架构合理性。
    输出格式：[技术评分: X/10] 评估内容""",
    output_key="tech_assessment"
)

business_expert = LlmAgent(
    name="BusinessExpert",
    model="gemini-2.0-flash",
    instruction="""你是一位商业分析师。
    请从商业价值角度评估以下产品方案（100字以内）：
    评估维度：市场潜力、盈利模式、竞争优势。
    输出格式：[商业评分: X/10] 评估内容""",
    output_key="business_assessment"
)

# 结果聚合 Agent
aggregator = LlmAgent(
    name="Aggregator",
    model="gemini-2.0-flash",
    instruction="""你是一位产品总监。
    请综合以下三份评估报告（来自 session state 的 ux_assessment、
    tech_assessment、business_assessment），给出综合评估结论和建议。
    输出：综合评分、主要风险、关键建议（各不超过3条）""",
    output_key="final_assessment"
)

# 并行评估 + 串行聚合
evaluation_pipeline = SequentialAgent(
    name="EvaluationPipeline",
    sub_agents=[
        ParallelAgent(
            name="ParallelEvaluation",
            sub_agents=[ux_expert, tech_expert, business_expert]
        ),
        aggregator
    ]
)

# 运行
session_service = InMemorySessionService()
runner = Runner(
    agent=evaluation_pipeline,
    session_service=session_service,
    app_name="eval_app"
)

session = session_service.create_session(app_name="eval_app", user_id="user_1")
product_proposal = """
产品方案：AI 驱动的个人财务管理助手
- 自动分类记账，学习用户消费习惯
- 智能预算建议，提前预警超支风险
- 投资组合优化建议（非执行）
- 目标用户：25-40岁城市白领
"""

runner.run(
    user_id="user_1",
    session_id=session.id,
    new_message=types.Content(parts=[types.Part(text=product_proposal)])
)

session_data = session_service.get_session(
    app_name="eval_app", user_id="user_1", session_id=session.id
)
print("=== 综合评估结果 ===")
print(session_data.state.get("final_assessment"))

"""
示例：智能客服分诊系统
基于用户意图路由到不同专业 Agent
"""

from google.adk.agents import LlmAgent
from google.adk.sessions import InMemorySessionService
from google.adk.runners import Runner
from google.genai import types

# 专业 Agent
tech_agent = LlmAgent(
    name="TechSupportAgent",
    model="gemini-2.0-flash",
    instruction="""你是技术支持专家。请解答用户的技术问题，
    提供具体的操作步骤和解决方案。"""
)

billing_agent = LlmAgent(
    name="BillingAgent",
    model="gemini-2.0-flash",
    instruction="""你是财务客服专员。请处理用户的账单、退款、
    订阅相关问题，提供准确的账单说明。"""
)

general_agent = LlmAgent(
    name="GeneralInfoAgent",
    model="gemini-2.0-flash",
    instruction="""你是通用客服代表。回答用户关于产品功能、
    政策和一般性问题。"""
)

# 路由 Coordinator
coordinator = LlmAgent(
    name="CustomerServiceCoordinator",
    model="gemini-2.0-flash",
    instruction="""你是智能客服调度中心。分析用户问题并路由到合适的专员：
    - 技术问题（安装、Bug、功能异常）→ TechSupportAgent
    - 账单问题（付款、退款、订阅）→ BillingAgent
    - 其他问题 → GeneralInfoAgent
    
    使用 transfer_to_agent 工具完成路由，不要自己回答用户问题。""",
    sub_agents=[tech_agent, billing_agent, general_agent]
)

"""
示例：深度竞品分析系统
Orchestrator 动态决定分析维度，Workers 并行执行
"""

from google.adk.agents import LlmAgent
from google.adk.tools import agent_tool
from google.adk.sessions import InMemorySessionService
from google.adk.runners import Runner
from google.genai import types

# Worker Agents（被封装为 AgentTool）
product_feature_analyst = LlmAgent(
    name="ProductFeatureAnalyst",
    model="gemini-2.0-flash",
    instruction="""你是产品功能分析专家。
    分析给定产品/公司的核心功能特性、差异化竞争点。
    输出结构化的功能分析报告。"""
)

pricing_analyst = LlmAgent(
    name="PricingAnalyst",
    model="gemini-2.0-flash",
    instruction="""你是定价策略专家。
    分析给定产品/公司的定价模型、价格梯度、目标客群定价策略。
    输出定价策略分析。"""
)

market_position_analyst = LlmAgent(
    name="MarketPositionAnalyst",
    model="gemini-2.0-flash",
    instruction="""你是市场定位专家。
    分析给定产品/公司在市场中的定位、品牌认知、用户口碑。
    输出市场定位分析。"""
)

tech_stack_analyst = LlmAgent(
    name="TechStackAnalyst",
    model="gemini-2.0-flash",
    instruction="""你是技术分析专家。
    分析给定产品/公司的技术架构、AI能力、技术护城河。
    输出技术栈分析。"""
)

# 将 Workers 封装为工具
feature_tool = agent_tool.AgentTool(agent=product_feature_analyst)
pricing_tool = agent_tool.AgentTool(agent=pricing_analyst)
market_tool = agent_tool.AgentTool(agent=market_position_analyst)
tech_tool = agent_tool.AgentTool(agent=tech_stack_analyst)

# Orchestrator Agent
orchestrator = LlmAgent(
    name="CompetitiveIntelligenceOrchestrator",
    model="gemini-2.0-flash",
    instruction="""你是竞争情报分析调度专家。
    
    当用户要求分析某个竞争对手时，你需要：
    1. 调用以下工具并行收集各维度分析（一次性全部调用以提高效率）：
       - ProductFeatureAnalyst: 产品功能分析
       - PricingAnalyst: 定价策略分析
       - MarketPositionAnalyst: 市场定位分析
       - TechStackAnalyst: 技术栈分析
    
    2. 综合所有分析，生成完整的竞品报告，包括：
       - 执行摘要（核心发现，3条）
       - 四维分析详情
       - 对我方的战略建议（2-3条）
    
    报告要有深度，避免泛泛而谈。""",
    tools=[feature_tool, pricing_tool, market_tool, tech_tool]
)

# 运行
session_service = InMemorySessionService()
runner = Runner(agent=orchestrator, session_service=session_service, app_name="ci_app")

session = session_service.create_session(app_name="ci_app", user_id="analyst_1")
runner.run(
    user_id="analyst_1",
    session_id=session.id,
    new_message=types.Content(parts=[types.Part(
        text="请对 Notion AI 进行全面的竞品分析"
    )])
)

"""
示例：代码质量迭代优化
Generator 生成代码，Critic 评审，循环优化直到满分
"""

from google.adk.agents import LoopAgent, LlmAgent
from google.adk.sessions import InMemorySessionService
from google.adk.runners import Runner
from google.genai import types

# Generator Agent
code_generator = LlmAgent(
    name="CodeGenerator",
    model="gemini-2.0-flash",
    instruction="""你是一位 Python 工程师。

    如果 session state 中没有 'code_draft'，
    根据用户需求生成初始代码实现。
    
    如果存在 'critic_feedback'，
    根据反馈改进 'code_draft' 中的代码。
    
    输出：只输出 Python 代码，不要解释。
    将代码存入 output_key 'code_draft'。""",
    output_key="code_draft"
)

# Critic/Evaluator Agent  
code_critic = LlmAgent(
    name="CodeCritic",
    model="gemini-2.0-flash",
    instruction="""你是一位严格的代码审查专家。
    
    审查 session state 中 'code_draft' 的代码，从以下维度评分（各25分）：
    1. 正确性：逻辑是否无误
    2. 可读性：命名、注释、结构
    3. 健壮性：错误处理、边界情况
    4. 性能：算法效率
    
    输出格式（严格遵守）：
    总分: XX/100
    
    问题清单:
    - [具体问题1]
    - [具体问题2]
    
    如果总分 >= 85，在最后一行输出：APPROVED
    
    将你的评审结果存入 output_key 'critic_feedback'。
    将 'APPROVED' 或 'NEEDS_REVISION' 存入 output_key 'review_status'。""",
    output_key="critic_feedback"  # 注：实际需要自定义 tool 写多个 key
)

# 终止条件检查 Agent
termination_checker = LlmAgent(
    name="TerminationChecker",
    model="gemini-2.0-flash",
    instruction="""检查 session state 中 'critic_feedback' 是否包含 'APPROVED'。
    如果包含 APPROVED，调用 escalate() 终止循环。
    否则，不做任何操作，让循环继续。""",
)

# 循环 Agent（最多5轮迭代）
optimization_loop = LoopAgent(
    name="CodeOptimizationLoop",
    sub_agents=[code_generator, code_critic, termination_checker],
    max_iterations=5
)

# 运行
session_service = InMemorySessionService()
runner = Runner(
    agent=optimization_loop,
    session_service=session_service,
    app_name="code_optimizer"
)

session = session_service.create_session(
    app_name="code_optimizer", user_id="dev_1"
)

runner.run(
    user_id="dev_1",
    session_id=session.id,
    new_message=types.Content(parts=[types.Part(
        text="""实现一个 LRU Cache 类，支持 get(key) 和 put(key, value) 操作，
        容量满时自动淘汰最久未使用的元素。时间复杂度要求 O(1)。"""
    )])
)

final_state = session_service.get_session(
    app_name="code_optimizer", user_id="dev_1", session_id=session.id
).state

print("=== 最终代码 ===")
print(final_state.get("code_draft"))
print("\n=== 评审结果 ===")
print(final_state.get("critic_feedback"))

"""
Human-in-the-Loop 示例：策略引擎驱动的智能审批系统
低风险操作自动执行，高风险操作暂停等待人工确认
"""

import asyncio
from dataclasses import dataclass
from enum import Enum
from typing import Any, Optional
from google.adk.agents import LlmAgent
from google.adk.tools import FunctionTool
from google.adk.sessions import InMemorySessionService
from google.adk.runners import Runner
from google.genai import types

# ===== 策略引擎 =====
class RiskLevel(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"

@dataclass
class PolicyDecision:
    risk_level: RiskLevel
    requires_approval: bool
    reason: str

class PolicyEngine:
    """基于规则的风险评估引擎"""
    
    RULES = {
        "delete": RiskLevel.HIGH,
        "send_email_batch": RiskLevel.HIGH,
        "update_production": RiskLevel.CRITICAL,
        "read_data": RiskLevel.LOW,
        "generate_report": RiskLevel.LOW,
        "send_notification": RiskLevel.MEDIUM,
        "update_config": RiskLevel.MEDIUM,
    }
    
    @classmethod
    def evaluate(cls, action: str, context: dict) -> PolicyDecision:
        base_risk = cls.RULES.get(action, RiskLevel.MEDIUM)
        
        # 上下文调整：影响人数越多，风险越高
        affected_users = context.get("affected_users", 0)
        if affected_users > 1000 and base_risk == RiskLevel.MEDIUM:
            base_risk = RiskLevel.HIGH
        
        requires_approval = base_risk in [RiskLevel.HIGH, RiskLevel.CRITICAL]
        
        return PolicyDecision(
            risk_level=base_risk,
            requires_approval=requires_approval,
            reason=f"操作'{action}'风险等级: {base_risk.value}"
        )

# ===== 人工确认模拟器 =====
class HumanApprovalGateway:
    """
    生产环境中，这里对接企业审批系统（钉钉、飞书、Slack）
    演示环境使用命令行交互
    """
    
    @staticmethod
    async def request_approval(
        action: str,
        details: dict,
        risk_level: str
    ) -> bool:
        print(f"\n{'='*50}")
        print(f"需要人工审核")
        print(f"操作: {action}")
        print(f"风险等级: {risk_level.upper()}")
        print(f"详情: {details}")
        print(f"{'='*50}")
        
        response = input("批准此操作? (yes/no): ").strip().lower()
        return response in ["yes", "y", "是", "批准"]

# ===== 工具定义 =====
approval_gateway = HumanApprovalGateway()
policy_engine = PolicyEngine()

async def execute_action(
    action: str,
    parameters: dict,
    description: str
) -> dict:
    """
    通用操作执行工具，内置策略引擎和人工审核
    
    Args:
        action: 操作类型（如 delete, send_email_batch, read_data）
        parameters: 操作参数
        description: 操作的人可读描述
    
    Returns:
        执行结果
    """
    # 策略评估
    decision = policy_engine.evaluate(action, parameters)
    
    if decision.risk_level == RiskLevel.CRITICAL:
        return {
            "status": "blocked",
            "reason": f"操作被策略引擎强制拒绝: {decision.reason}"
        }
    
    if decision.requires_approval:
        # 请求人工审核
        approved = await approval_gateway.request_approval(
            action=action,
            details={"description": description, "parameters": parameters},
            risk_level=decision.risk_level.value
        )
        
        if not approved:
            return {
                "status": "rejected",
                "reason": "操作被人工审核拒绝"
            }
    
    # 模拟执行（实际场景中调用真实 API）
    print(f"执行操作: {description}")
    return {
        "status": "success",
        "action": action,
        "message": f"操作'{description}'已成功执行",
        "auto_approved": not decision.requires_approval
    }

# ===== Agent 配置 =====
secure_agent = LlmAgent(
    name="SecureOperationsAgent",
    model="gemini-2.0-flash",
    instruction="""你是一位企业 IT 运营助手。
    
    当用户要求执行操作时，使用 execute_action 工具来完成。
    
    操作类型参考：
    - read_data: 读取数据
    - generate_report: 生成报告
    - send_notification: 发送通知
    - update_config: 更新配置
    - delete: 删除数据
    - send_email_batch: 批量发送邮件
    - update_production: 更新生产环境
    
    始终提供清晰的 description 说明操作目的和影响范围。""",
    tools=[FunctionTool(execute_action)]
)

# ===== 运行示例 =====
async def main():
    session_service = InMemorySessionService()
    runner = Runner(
        agent=secure_agent,
        session_service=session_service,
        app_name="secure_ops"
    )
    
    session = session_service.create_session(
        app_name="secure_ops", user_id="ops_1"
    )
    
    # 测试不同风险级别的操作
    test_requests = [
        "生成过去30天的用户活跃度报告",  # LOW risk，自动执行
        "向2000名用户发送产品更新通知邮件",  # HIGH risk，需要审核
    ]
    
    for request in test_requests:
        print(f"\n{'='*60}")
        print(f"用户请求: {request}")
        
        async for event in runner.run_async(
            user_id="ops_1",
            session_id=session.id,
            new_message=types.Content(parts=[types.Part(text=request)])
        ):
            if event.is_final_response():
                print(f"Agent 回复: {event.content.parts[0].text}")

asyncio.run(main())

from google.adk.agents import SequentialAgent, ParallelAgent, LoopAgent, LlmAgent

# 构建嵌套多智能体系统
research_pipeline = SequentialAgent(
    name="ResearchPipeline",
    sub_agents=[
        # 第1阶段：并行信息收集
        ParallelAgent(
            name="InformationGathering",
            sub_agents=[web_search_agent, db_query_agent, expert_agent]
        ),
        # 第2阶段：信息综合
        synthesis_agent,
        # 第3阶段：迭代优化
        LoopAgent(
            name="QualityOptimization",
            sub_agents=[draft_agent, critic_agent],
            max_iterations=3
        ),
        # 第4阶段：人工审核（仅对外发布报告）
        human_review_agent
    ]
)

# 糟糕的状态设计
session.state["data"] = "..."  # key 太模糊
session.state["result"] = "..."  # 会被多个 Agent 覆盖

# 良好的状态设计
session.state["pipeline:step1:raw_data"] = "..."      # 命名空间隔离
session.state["pipeline:step2:analysis"] = "..."      # 阶段清晰
session.state["pipeline:final:report"] = "..."        # 最终输出标记
session.state["pipeline:metadata:iteration_count"] = 3  # 元数据分离

import asyncio
from typing import Optional

class RobustAgentRunner:
    """健壮的 Agent 运行器，包含超时、重试、熔断"""
    
    def __init__(self, agent, max_retries: int = 3, timeout_seconds: int = 60):
        self.agent = agent
        self.max_retries = max_retries
        self.timeout = timeout_seconds
    
    async def run_with_resilience(
        self, 
        input_data: str,
        fallback_response: Optional[str] = None
    ) -> str:
        last_error = None
        
        for attempt in range(self.max_retries):
            try:
                result = await asyncio.wait_for(
                    self.agent.run_async(input_data),
                    timeout=self.timeout
                )
                return result
                
            except asyncio.TimeoutError:
                last_error = f"超时（{self.timeout}s），尝试 {attempt + 1}/{self.max_retries}"
                print(f"{last_error}")
                await asyncio.sleep(2 ** attempt)  # 指数退避
                
            except Exception as e:
                last_error = f"错误: {str(e)}"
                print(f"{last_error}")
                
                # 非可重试错误直接返回
                if "INVALID_ARGUMENT" in str(e):
                    break
        
        # 所有重试失败
        if fallback_response:
            return fallback_response
        raise RuntimeError(f"Agent 执行失败（{self.max_retries}次重试后）: {last_error}")

# ADK 内置的事件流可观测性
async def run_with_full_observability(runner, user_id, session_id, message):
    """带完整事件追踪的 Agent 运行"""
    
    event_log = []
    
    async for event in runner.run_async(
        user_id=user_id,
        session_id=session_id,
        new_message=message
    ):
        # 记录每个事件
        event_info = {
            "type": event.type,
            "agent": getattr(event, "author", "unknown"),
            "timestamp": asyncio.get_event_loop().time()
        }
        
        if event.content:
            # 工具调用事件
            for part in event.content.parts:
                if hasattr(part, "function_call"):
                    event_info["tool_call"] = {
                        "name": part.function_call.name,
                        "args": dict(part.function_call.args)
                    }
                    print(f"🔧 [{event_info['agent']}] 调用工具: {part.function_call.name}")
                
                elif hasattr(part, "function_response"):
                    event_info["tool_response"] = str(part.function_response.response)[:200]
                    print(f"工具响应: {str(part.function_response.response)[:100]}...")
                
                elif hasattr(part, "text") and part.text:
                    print(f"[{event_info['agent']}]: {part.text[:200]}")
        
        event_log.append(event_info)
        
        if event.is_final_response():
            print(f"\n任务完成，共 {len(event_log)} 个事件")
            break
    
    return event_log

# 使用 LangSmith 或 Langfuse 进行生产级追踪
# pip install langfuse
from langfuse import Langfuse

langfuse = Langfuse(
    secret_key="your-secret-key",
    public_key="your-public-key", 
    host="https://cloud.langfuse.com"
)

trace = langfuse.trace(name="multi_agent_pipeline")
span = trace.span(name="orchestrator_agent")
# ... 在 Agent 调用前后记录 span

# 生产级配置示例
PRODUCTION_CONFIG = {
    "max_retries": 3,
    "timeout_per_agent": 30,  # seconds
    "max_pipeline_time": 300,  # 5 minutes
    "cost_limit_per_request": 0.50,  # USD
    "fallback_enabled": True,
    "human_escalation_threshold": "CRITICAL",
    "observability": {
        "tracing": True,
        "metrics": True,
        "log_level": "INFO"
    }
}