多Agent交接断层六栈实测：上下文丢失到结构化移交全拆解

graph TB
    subgraph "交接层架构"
        direction TB
        A[Agent A 输出] --> B{交接协议层}
        B -->|结构化序列化| C[State Schema]
        B -->|摘要压缩| D[Context Window]
        B -->|协议封装| E[A2A/Handoff]

        C --> F{状态校验}
        D --> G{保真度检查}
        E --> H{协议解析}

        F --> I[Agent B 接收]
        G --> I
        H --> I

        I --> J{上下文完整?}
        J -->|是| K[正常执行]
        J -->|否| L[降级/回滚]
        L --> M[Checkpoint恢复]
        L --> N[人工介入]
    end

    style B fill:#ff6b6b,color:#fff
    style J fill:#ffd93d,color:#333
    style K fill:#6bcb77,color:#fff
    style L fill:#ff8c42,color:#fff

from langgraph.graph import StateGraph, MessagesState, START
from langgraph.graph.message import add_messages
from typing import Annotated, TypedDict

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]  # 累积追加，不丢失
    current_plan: str                         # 取最新
    attempted_solutions: Annotated[list, lambda x,y: x+y]  # 只增不减
    constraints: dict                         # 合并而非覆盖
    confidence: float                         # 取最新

def agent_a(state: AgentState) -> dict:
    # Agent A产出结构化状态
    return {
        "current_plan": "搜索2026年Q1财报数据",
        "attempted_solutions": ["已尝试SEC EDGAR API"],
        "constraints": {"language": "zh-CN", "no_external_api": True},
        "confidence": 0.85
    }

def agent_b(state: AgentState) -> dict:
    # Agent B收到完整状态，能看到所有历史尝试和约束
    attempted = state["attempted_solutions"]  # 包含A的所有尝试
    constraints = state["constraints"]        # 包含A的所有约束
    # 不会重复A已失败的方案，不会违反A设定的约束
    return {"current_plan": "从本地缓存提取财报数据"}

graph = StateGraph(AgentState)
graph.add_node("researcher", agent_a)
graph.add_node("analyst", agent_b)
graph.add_edge("researcher", "analyst")

from crewai import Agent, Task, Crew, Process
from crewai.memory import LongTermMemory

researcher = Agent(
    role="研究员",
    goal="收集并验证数据",
    backstory="你是严谨的数据研究员",
    allow_delegation=False
)

analyst = Agent(
    role="分析师",
    goal="基于数据生成洞察",
    backstory="你是资深行业分析师",
    allow_delegation=False
)

# 关键：context参数控制交接内容
research_task = Task(
    description="研究2026年AI Agent市场格局",
    agent=researcher,
    expected_output="包含数据来源的结构化研究报告"
)

# analyst_task自动接收research_task的输出
analyst_task = Task(
    description="基于研究结果生成投资建议",
    agent=analyst,
    context=[research_task],  # 显式声明依赖
    expected_output="投资建议报告"
)

crew = Crew(
    agents=[researcher, analyst],
    tasks=[research_task, analyst_task],
    process=Process.sequential,
    memory=True  # 启用记忆系统辅助交接
)

from agents import Agent, Runner, handoff, RunContextWrapper
from agents.extensions import handoff_filters

# 定义专业Agent
billing_agent = Agent(
    name="账单专家",
    instructions="处理账单查询和争议。严格使用用户提供的数据。"
)

refund_agent = Agent(
    name="退款专员", 
    instructions="处理退款请求。必须确认原始购买记录。"
)

# 主路由Agent使用handoff工具
triage_agent = Agent(
    name="路由专员",
    instructions="判断用户问题类型并转交专业Agent。",
    handoffs=[
        handoff(
            agent=billing_agent,
            # 关键：input_filter控制交接时传递什么上下文
            input_filter=handoff_filters.remove_all_tools,
        ),
        handoff(
            agent=refund_agent,
            input_filter=handoff_filters.truncate_history(max_tokens=2000),
        ),
    ]
)

# 上下文类型控制：结构化数据不在对话历史中丢失
from pydantic import BaseModel

class UserContext(BaseModel):
    user_id: str
    purchase_history: list[dict]
    complaint_count: int

result = Runner.run_sync(
    triage_agent, 
    "我要退款",
    context=UserContext(
        user_id="u_12345",
        purchase_history=[{"id": "p_001", "amount": 299}],
        complaint_count=2
    )
)

import autogen
from autogen import GroupChat, GroupChatManager

# 定义专业Agent
researcher = autogen.AssistantAgent(
    name="Researcher",
    system_message="你是研究Agent。完成研究后说TERMINATE或转交。",
    llm_config={"config_list": [{"model": "gpt-4o"}]},
    max_consecutive_auto_reply=3
)

coder = autogen.AssistantAgent(
    name="Coder",
    system_message="你是编程Agent。只根据Researcher提供的数据编写代码。",
    llm_config={"config_list": [{"model": "gpt-4o"}]}
)

# GroupChat管理交接：Selector决定谁发言
groupchat = GroupChat(
    agents=[researcher, coder],
    messages=[],
    max_round=10,
    # 自定义选择器：基于内容决定下一步
    select_speaker_prompt_template="""基于当前对话历史，判断哪个Agent应该发言。
    当前阶段: {current_stage}
    已完成任务: {completed_tasks}
    选择: """
)

manager = GroupChatManager(
    groupchat=groupchat,
    llm_config={"config_list": [{"model": "gpt-4o"}]}
)

# 发起对话，交接由Manager自动管理
user_proxy = autogen.UserProxyAgent(
    name="User", human_input_mode="NEVER"
)
user_proxy.initiate_chat(manager, message="研究并实现一个排序算法")

from google.adk import Agent
from google.adk.tools import a2a_tool

# Agent Card：声明能力，供其他Agent发现
research_agent = Agent(
    name="research_agent",
    model="gemini-2.5-pro",
    instruction="你是研究专家。",
    tools=[a2a_tool(
        agent_url="https://analytics-agent.example.com",
        # A2A协议自动处理：
        # 1. 发现对方AgentCard
        # 2. 创建Task并传递
        # 3. 等待对方Agent返回结果
        # 4. 将结果合并到当前上下文
    )]
)

# A2A Task生命周期：确保交接状态可追踪
# Task States: submitted → working → completed/failed/canceled
# 每次状态变更都有Artifact和Message记录

{
  "jsonrpc": "2.0",
  "method": "tasks/send",
  "params": {
    "id": "task_20260601_001",
    "sessionId": "session_abc",
    "message": {
      "role": "user",
      "parts": [
        {"type": "text", "text": "分析Q1财报数据"},
        {"type": "data", "data": {"quarter": "Q1", "year": 2026}}
      ]
    },
    "metadata": {
      "parent_task_id": "parent_task_001",
      "handoff_reason": "需要财务分析专业能力",
      "context_window_budget": 4000
    }
  }
}

// C# 示例：Semantic Kernel Process Framework
var process = new ProcessBuilder("DocumentPipeline")
    .AddStepFromType<ResearchAgent>("Researcher")
    .AddStepFromType<AnalysisAgent>("Analyst")
    .AddStepFromType<SummaryAgent>("Summarizer")
    // 关键：事件映射定义交接路径
    .OnEvent("ResearchCompleted")
        .SendEventTo("Analyst")
    .OnEvent("AnalysisCompleted") 
        .SendEventTo("Summarizer")
    .OnEvent("AnalysisFailed")
        .SendEventTo("Researcher")  // 失败回退路径
    .Build();

// Agent实现：emit event触发交接
public class ResearchAgent : KernelProcessStep
{
    [KernelFunction]
    public async Task ResearchAsync(
        KernelProcessStepContext context,  // 用于emit事件
        string topic,
        Dictionary<string, object> state  // 结构化状态传递
    )
    {
        var result = await DoResearch(topic);
        // 交接 = emit event + 携带结构化数据
        await context.EmitEventAsync(new KernelProcessEvent
        {
            EventId = "ResearchCompleted",
            Data = new ResearchOutput
            {
                Findings = result.Findings,
                Sources = result.Sources,
                ConfidenceScore = result.Confidence,
                // 结构化状态完整传递
                AttemptedApproaches = state.TryGetValue("attempts", out var a) 
                    ? (List<string>)a : new List<string>()
            }
        });
    }
}

graph TD
    A[多Agent交接需求] --> B{需要跨框架互操作?}
    B -->|是| C[Google ADK + A2A]
    B -->|否| D{交接复杂度?}

    D -->|简单线性| E{团队技术栈?}
    D -->|复杂DAG| F{需要失败回滚?}
    D -->|动态路由| G{绑定OpenAI生态?}

    E -->|Python优先| H{追求速度?}
    E -->|.NET生态| I[Semantic Kernel]

    H -->|快速原型| J[CrewAI]
    H -->|生产就绪| K[OpenAI Agents SDK]

    F -->|必须回滚| L[LangGraph]
    F -->|可接受重跑| M[OpenAI Agents SDK]

    G -->|是| N[OpenAI Agents SDK]
    G -->|否| O{需要多轮协商?}

    O -->|是| P[AG2 GroupChat]
    O -->|否| Q[LangGraph]

    C --> R[最终推荐]
    I --> R
    J --> R
    K --> R
    L --> R
    M --> R
    N --> R
    O --> R
    P --> R
    Q --> R

    style C fill:#4285f4,color:#fff
    style L fill:#1a1a2e,color:#fff
    style N fill:#10a37f,color:#fff