使用 Langchain 和 LangGraph 进行 Agent-to-Agent (A2A) 开发全攻略

引言：拥抱 Agent-to-Agent 的协作智能

随着大型语言模型（LLM）能力的飞速发展，构建能够执行复杂任务的智能化系统已成为可能。单个 LLM Agent 擅长处理特定领域的任务，但在面对跨领域、需要多步骤协作或涉及不同专业知识的问题时，单一 Agent 的局限性便显现出来。Agent-to-Agent (A2A) 系统应运而生，它通过多个 Agent 之间的协同、沟通与合作，能够模拟人类团队协作的方式，分解复杂任务、整合多方信息、甚至进行谈判与决策，从而解决单个 Agent 无法完成的难题。

Langchain 作为 LLM 应用开发领域的流行框架，为构建 Agent 提供了强大的基础能力，包括与各种 LLM、工具和数据源的集成。而 LangGraph，作为 Langchain 的一个重要扩展，专注于有状态的、长期运行的 Agent 工作流编排，天然适合构建和管理复杂的多 Agent 协作系统。

本教程旨在为开发者提供一份详尽的 A2A 开发指南，重点讲解如何利用 Langchain 构建基础 Agent，并借助 LangGraph 实现 Agent 之间的通信、状态管理与协作流程编排，最终构建出强大的多 Agent 协作系统。我们将深入剖析 LangGraph 的核心概念，提供丰富的代码示例，覆盖从环境搭建到复杂 A2A 架构模式的实现，帮助您掌握 A2A 开发的关键技术，并将所学应用于实际项目。

"多 Agent 系统通过分解复杂问题、整合多元专业知识，能够解决单个 Agent 力所不能及的任务，是通向更高级人工智能应用的关键一步。"

第一章：A2A 系统与 Langchain/LangGraph 的基础

1.1 什么是 Agent？多 Agent 系统？A2A 通信？

• Agent (智能体)： 在 LLM 应用开发领域，Agent 是指能够利用 LLM 作为其推理核心，并根据当前状态和目标，动态决定并执行一系列动作（调用工具、与环境交互等）以达成目标的系统。与传统的硬编码执行流程不同，Agent 的行为更具灵活性和自主性，能够根据上下文进行规划、学习和适应。 Langchain Agent 的核心在于使用 LLM 作为决策引擎，通过提示工程和工具调用实现动态行为。

  例如，一个“研究 Agent”可能接收一个主题，然后自己决定什么时候去搜索网页、什么时候去阅读文档、什么时候去提取关键信息，这些动作不是预先设定好的固定流程，而是由其内置的 LLM 根据情况判断并执行的。

• 多 Agent 系统 (Multi-Agent System, MAS)： 多 Agent 系统是由两个或多个 Agent 组成的集合，这些 Agent 能够相互影响、协作或竞争，共同完成一个或一系列任务。每个 Agent 可能拥有特定的能力、知识或目标，通过分工与协作，整个系统可以展现出比单个 Agent 更高级和复杂的能力。

  多 Agent 系统的优势包括：

  • 模块化：将复杂任务分解给多个 Agent，降低开发和维护复杂度。

  • 专业化：每个 Agent 可以专注于特定领域或任务，提高效率和性能。

  • 鲁棒性：某个 Agent 失败时，其他 Agent 可能可以接管或弥补。

  • 分布式：Agent 可以部署在不同的环境中，实现分布式计算和协作。

  一个简单的例子：在一个客服系统中，可能有一个“意图识别 Agent”负责理解用户问题，一个“常见问题解答 Agent”处理标准问题，一个“技术支持 Agent”处理复杂技术问题，以及一个“财务 Agent”处理支付问题。这些 Agent 协同工作，共同为用户提供支持。

• Agent-to-Agent (A2A) 通信： 在多 Agent 系统中，A2A 通信是指 Agent 之间进行信息交换、状态同步、任务分配和协调行为的过程。有效的 A2A 通信是实现 Agent 协同工作的关键。 Agent 之间可能通过共享内存、消息队列、API 调用或者更高级的 Agent 协议进行通信。

  A2A 通信模式多样，例如：

  • 消息传递： Agent 发送和接收消息，消息可能包含数据、指令或状态更新。

  • 状态共享： Agent 访问和修改共享的全局或部分状态，实现信息协同。

  • 任务委托：一个 Agent 将子任务分配给另一个 Agent 执行。

  • 协商与拍卖： Agent 之间通过协商或拍卖机制分配资源或任务。

  在 LangGraph 中，A2A 通信主要通过状态共享（修改共享状态）和Handoffs（控制权移交）来实现。

1.2 Langchain 在 Agent 开发中的角色与优势

Langchain 为构建 LLM 驱动的 Agent 提供了标准化的接口和丰富的组件：

• 模型集成：轻松接入各种 LLM 提供商（OpenAI, Google, Anthropic, 本地模型等）。

• 工具抽象：将外部服务（搜索引擎、数据库、API 等）封装为 Agent 可调用的工具。提供了大量的预置工具，并支持自定义工具。

• 提示工程：提供了灵活的提示模板，方便构建 Agent 的系统指令、少样本示例和 ReAct 推理格式。

• AgentExecutor：用于运行 Agent，管理 LLM 和工具之间的交互循环，处理错误和日志。

• Memory/State Management：提供多种方法管理 Agent 的会话历史和长期记忆，为 A2A 系统提供状态管理基础。

通过 Langchain，开发者可以快速构建一个能够感知、推理、行动的基础 Agent，为后续的多 Agent 协作奠定基础。

1.3 LangGraph 在 A2A 开发中的角色与优势

LangGraph 是在 Langchain 之上构建的库，专门用于构建有状态的、基于图形的 Agent 和多 Agent 系统。它解决了单个 Agent 或简单链式结构在处理复杂、多步、有状态任务时的不足。

LangGraph 的核心优势在于：

• 基于图的编排：使用图（Graph）结构清晰地定义 Agent（节点）和它们之间的交互流程（边）。这使得复杂工作流的可视化和管理变得容易。

• 状态管理：内置强大的状态管理能力，Agent 之间通过共享状态进行通信和协作。状态可以持久化，支持长期运行和中断恢复。

• 灵活的控制流：支持条件分支、循环、子图等复杂的控制流逻辑，能够根据不同的事件或状态变化动态地路由任务给不同的 Agent。

• 持久执行与人机协作 (Human-in-the-Loop)：支持复杂Agent的持久执行，即使系统中断也能从中断点恢复。可以方便地在图的任意节点插入人类干预点。

• Agent 定义与集成：LangGraph 可以轻松集成 Langchain Agent 作为图中的节点，或者直接定义节点函数实现 Agent 的部分逻辑。

• 调试与可视化：通过与 LangSmith 的集成，可以方便地追踪 Agent 系统的执行流程、状态变化和 Agent 间的通信，极大地简化了调试。

对于 A2A 开发，LangGraph 提供了理想的框架来：

• 定义不同的 Agent 节点，每个节点负责特定的任务或领域。

• 设计 Agent 之间的通信通道，通过共享状态传递信息。

• 编排 Agent 的协作流程，决定任务如何从一个 Agent 传递到另一个 Agent。

• 管理整个 Agent 系统的状态，确保信息在 Agent 之间的一致性和可追溯性。

可以说，Langchain 提供了构建 Agent 的“砖瓦”，而 LangGraph 提供了将这些“砖瓦”搭建成复杂多层建筑的“蓝图”和“骨架”，特别适合构建需要 Agent 之间紧密协作和状态共享的 A2A 系统。

第二章：开发环境搭建

2.1 安装 Python 和必要的库

首先，确保您安装了 Python 3.8 或更高版本。推荐使用虚拟环境来管理项目依赖。

创建一个虚拟环境（例如使用 venv）：

python -m venv .venv

激活虚拟环境：

• 在 macOS 和 Linux 上：

  source .venv/bin/activate

• 在 Windows 上：

  .venv\Scripts\activate

激活虚拟环境后，安装 Langchain、LangGraph 以及一些常用的依赖库：

pip install langchain langchain-core langchain-community langgraph jupyterhub

2.2 配置 LLM 访问权限

Langchain 和 LangGraph 需要使用 LLM。您需要根据选择的 LLM 提供商进行相应的配置。以 OpenAI 为例，您需要设置 OPENAI_API_KEY 环境变量：

export OPENAI_API_KEY="您的 OpenAI API 密钥"

如果您使用 Google 的 Gemini 模型（通过 Vertex AI），需要配置 Google Cloud 项目并进行身份认证。具体步骤参考 Google Cloud Vertex AI 文档²。

对于其他 LLM 提供商（如 Anthropic、Mistral 等），请查阅相应的 Langchain 文档了解如何设置 API 密钥或连接本地模型。

2.3 （可选）配置 LangSmith 进行调试和追踪

LangSmith 是 Langchain 提供的开发者平台，用于调试、测试和监控 LLM 应用程序。它与 LangGraph 深度集成，可以可视化 Agent 系统的执行图和状态变化，极大地提高了 A2A 系统的调试效率。强烈推荐配置 LangSmith。

注册 LangSmith 账号后，获取您的 API 密钥并设置环境变量：

export LANGCHAIN_TRACING_V2="true"
export LANGCHAIN_API_KEY="您的 LangSmith API 密钥"
export LANGCHAIN_PROJECT="您的 LangSmith 项目名称 (例如: A2A-Tutorial)"

设置这些环境变量后，Langchain 和 LangGraph 的运行痕迹将自动发送到 LangSmith 平台，您可以在 LangSmith UI 中查看详细的执行流程和状态变化。

2.4 验证安装

安装完成后，可以编写一个简单的 Python 脚本来验证安装：

from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI

# 确保您已设置 OPENAI_API_KEY 环境变量
llm = ChatOpenAI(model="gpt-4o-mini") # 选择您有权限访问的模型

try:
    response = llm.invoke([HumanMessage(content="Hello, LLM!")])
    print("LLM Response:", response.content)

    from langgraph.graph import StateGraph

    print("LangGraph installed successfully.")

except Exception as e:
    print("Error:", e)
    print("请检查您的库是否正确安装，API 密钥是否已配置。")

运行此脚本，如果能看到 LLM 的回复并且没有错误信息，说明环境已成功搭建。

第三章：LangGraph 核心概念深入剖析

LangGraph 的核心是利用有向无环图 (DAG) 或有向图 (DG) 来定义 Agent 工作流。图中的每个 Agent 或处理步骤表示为一个节点 (Node)，节点之间的连接表示为边 (Edge)。整个图有一个状态 (State)，在节点之间共享并更新。

3.1 图 (Graph) 的基本构成：节点 (Nodes) 和边 (Edges)

• 节点 (Node)： 在 LangGraph 中，节点代表图中的一个处理步骤或 Agent。一个节点可以是一个函数、一个 Langchain Runnable（如 LLM 调用、工具调用、另一个链或 AgentExecutor）或一个 LangGraph 子图。当图的执行流进入一个节点时，节点的功能被执行，并可能产生输出。

  节点函数的签名通常接受当前的图状态作为输入，并返回一个用于更新状态的值。

  例如，一个节点可以是：

  • 调用某个 LLM Agent 来生成回复。

  • 执行一个工具调用（如搜索、计算）。

  • 对当前状态进行某种转换或判断。

  • 调用另一个子图来处理更复杂的任务。

• 边 (Edge)： 边连接图中的节点，定义了执行流的路径。边可以是：

  • 普通边 (Normal Edge)：从一个节点到另一个节点的简单转移。当前一个节点执行完成后，执行流无条件地转移到由边指向的下一个节点。

  • 条件边 (Conditional Edge)：从一个节点根据其执行结果有条件地转移到多个可能的下一个节点中的一个。这是实现复杂逻辑判断和流程分支的关键。

  LangGraph 的图定义通常包括定义节点和定义边（以及入口点和可能的循环）。

3.2 状态管理 (State)

状态是 LangGraph 的核心概念之一。整个图的执行过程共享一个状态对象。每个节点在执行时都可以访问当前状态，并返回一个值来更新状态。LangGraph 会将节点返回的值与当前状态合并（默认是字典的 update 操作），形成新的状态，供下一个节点访问。

• 定义状态 (State Schema)： 状态通常定义为一个 Python TypedDict 或普通的 dict。TypedDict 提供了类型提示，有助于代码的清晰性和健壮性。状态包含了整个工作流中共享的所有信息。

  例如，一个简单的状态可以包含消息列表：

  from typing import TypedDict, Annotated, List
  from langchain_core.messages import BaseMessage
  import operator
  
  class AgentState(TypedDict):
      # `messages` 是一个包含了 BaseMessage 对象的列表
      # `operator.add` 是一个函数，用于将新的消息添加到列表中
      messages: Annotated[List[BaseMessage], operator.add]
      # 可能包含其他状态信息，例如：
      # search_results: str
      # task_status: str

  这里的 Annotated 和 operator.add 是 LangGraph 用于定义状态更新方式的一种机制。Annotated[List[BaseMessage], operator.add] 表示 messages 字段是一个 BaseMessage 列表，当一个节点返回一个列表来更新状态时，LangGraph 会使用 operator.add (即列表拼接) 将返回的列表添加到当前的 messages 列表中。对于简单的字段，如字符串或整数，直接指定类型即可，默认更新方式是覆盖（如果返回非 None 值）。

• 更新状态： 节点函数通过返回一个与 State Schema 兼容的字典来更新状态。LangGraph 负责合并过程。

  def call_llm(state: AgentState) -> dict:
      messages = state['messages']
      # 调用 LLM 处理 messages
      response = llm.invoke(messages)
      # 返回一个字典更新状态，将 LLM 的回复添加到 messages 列表
      return {"messages": [response]}

• 持久化状态 (Checkpointing)： LangGraph 支持状态的持久化，这意味着工作流可以在执行过程中保存当前状态，并在需要时从保存的状态恢复执行。这对于构建长期运行、可中断的 A2A 系统至关重要。通过配置检查点 (checkpoint) 后端（例如内存、SQLite、数据库），可以在图执行时自动保存状态。

  from langgraph.checkpoint.sqlite import SqliteSaver
  
  memory = SqliteSaver.from_conn_string(":memory:") # 使用内存作为示例，也可以指定文件路径
  
  # 编译图时传入检查点配置
  app = graph.compile(checkpointer=memory)
  
  # 调用时指定 config，其中 thread_id 用于标识会话
  config = {"configurable": {"thread_id": "user-123"}}
  result = app.invoke({"messages": [HumanMessage(content="Hello!")]}, config=config)
  
  # 稍后可以恢复该会话的状态
  restored_state = memory.get({"configurable": {"thread_id": "user-123"}})
  

  持久化状态是 enabling A2A 系统长期记忆和跨会话协作的关键。

3.3 条件边 (Conditional Edges)、入口点 (Entry Point)、循环 (Cycles)

• 条件边 (Conditional Edges)： 条件边是 LangGraph 实现动态流程控制的核心机制。从一个节点出发的条件边需要指定一个判断函数。这个判断函数接收当前状态作为输入，并返回一个字符串，表示下一个应该转移到的节点的名称。

  def should_continue(state: AgentState) -> str:
      messages = state['messages']
      last_message = messages[-1]
      # 根据最后一个消息的内容或类型判断下一步
      if "结束" in last_message.content:
          return "end" # 返回一个字符串，对应图中的一个节点或特殊的 END 节点
      else:
          return "continue"

  定义条件边时，需要：

  1. 指定起始节点 (source node)。

  2. 指定判断函数 (condition function)。

  3. 指定一个字典，将判断函数的返回结果映射到下一个节点 (mapping)。

  workflow.add_conditional_edges(
      node_a,           # 从 node_a 出发
      should_continue,  # 使用 should_continue 函数判断
      {                 # 映射关系
          "continue": node_b, # 如果返回 "continue"，转移到 node_b
          "end": END          # 如果返回 "end"，转移到特殊节点 END (终止执行)
      }
  )

• 入口点 (Entry Point)： 图的执行总需要一个起点。LangGraph 使用特殊的 START 符号来标识图的入口点。您需要使用 set_entry_point(node_name) 方法指定图的第一个执行节点。执行将从 START 节点开始，然后通过边转移到指定的入口节点。

  workflow.set_entry_point("first_node") # 设置 "first_node" 为图的入口
  workflow.add_edge(START, "first_node") # 通常会添加一条从 START 到入口节点的边

• 循环 (Cycles)： LangGraph 支持在图结构中存在循环。当执行流通过边回到已经访问过的节点时，就会形成循环。循环在 Agent 系统中非常常见，例如：

  • ReAct 模式中的“思考-行动-观察”循环。

  • 多轮对话中 Agent 不断处理用户输入并生成回复。

  • Agent 在未达到目标前反复尝试调用工具或与其他 Agent 交互。

  通过条件边，可以控制何时跳出循环（例如，达到某个条件、获取到足够信息、用户指示结束等）。LangGraph 的状态管理使得在循环中保持上下文信息成为可能。

3.4 特殊节点：END

END 是 LangGraph 中一个特殊的内置节点，表示图的执行终止。当执行流到达 END 节点时，整个 Agent 工作流结束。通常通过条件边将某些终止条件导向 END 节点。

第四章：一个基础的 LangGraph Agent 示例

本章我们将构建一个简单的 LangGraph Agent，它能够接收用户输入，调用 LLM 生成回复，并根据 LLM 的回复决定是否结束对话。这展示了 LangGraph 的基本结构、状态管理和条件分支。

import operator
from typing import TypedDict, Annotated, List

from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langchain_core.tools import tool # 导入 tool 装饰器用于简化工具定义
from langchain_openai import ChatOpenAI

from langgraph.graph import StateGraph, END, START # 导入 START 和 END 符号
from langgraph.checkpoint.sqlite import SqliteSaver # 用于状态持久化（可选）

# 1. 定义 AgentState：共享状态的 Schema
# AgentState 继承 TypedDict 表示类型化的字典
# Annotated[List[BaseMessage], operator.add] 表示 messages 是 BaseMessage 列表
# operator.add 是合并函数，用于将新返回的消息列表拼接到现有列表
class AgentState(TypedDict):
    messages: Annotated[List[BaseMessage], operator.add]
    # 可以添加其他状态字段，例如：
    # conversation_complete: bool

# 2. 定义节点函数

# 节点函数1: 调用 LLM
def call_llm(state: AgentState) -> dict:
    """
    调用 LLM 处理当前消息列表，并返回 LLM 的回复以更新状态。
    """
    messages = state['messages']
    print("--- Calling LLM ---")
    # 这里的 llm 可以是任何朗链集成的 ChatModel
    llm = ChatOpenAI(model="gpt-4o-mini")
    try:
        response = llm.invoke(messages)
        print(f"LLM Response: {response.content[:50]}...") # 打印部分回复方便查看
        # 返回字典，键与 AgentState 字段对应，值就是要添加或更新的数据
        return {"messages": [response]}
    except Exception as e:
        print(f"LLM Call Error: {e}")
        # 可以在这里处理错误，例如返回一个错误消息或者标记状态
        return {"messages": [AIMessage(content=f"抱歉，LLM 调用失败：{e}")]}


# 节点函数2: 判断对话是否结束 (条件判断函数)
def should_continue(state: AgentState) -> str:
    """
    根据 LLM 的最新回复判断对话是否应该继续。
    返回字符串来指导条件边跳转。
    """
    messages = state['messages']
    last_message = messages[-1] # 获取最新的 LLM 回复 (由 call_llm 添加到状态)
    print("--- Checking Conversation End ---")

    # 示例判断逻辑：如果最新回复是 AIMessage 并且包含特定关键词，则结束
    # 实际应用中，判断逻辑可能更复杂，例如分析用户意图、检查任务完成状态等
    if isinstance(last_message, AIMessage) and "再见" in last_message.content.lower():
        print("Conversation ends.")
        return "end"
    elif isinstance(last_message, AIMessage) and "完成" in last_message.content.lower():
         print("Conversation ends (task complete).")
         return "end"
    else:
        print("Conversation continues.")
        return "continue"

# 3. 定义图结构

# 创建 StateGraph 实例，传入状态 Type
workflow = StateGraph(AgentState)

# 添加节点：将节点函数添加到图中，并命名
workflow.add_node("call_llm", call_llm)

# 设置入口点：对话从调用 LLM 开始
workflow.set_entry_point("call_llm")

# 添加条件边：从 call_llm 节点出发，根据 should_continue 函数的结果跳转
workflow.add_conditional_edges(
    "call_llm",       # 条件边的起始节点
    should_continue,  # 条件判断函数
    {                 # 映射：判断函数的返回值到下一个节点
        "continue": "call_llm", # 如果返回 "continue"，继续回到 call_llm 形成循环
        "end": END              # 如果返回 "end"，终止执行 (到达 END 节点)
    }
)

# 4. 编译图：生成可执行的 LangGraph 应用
# checkpointer 用于持久化状态，这里使用内存存储
memory = SqliteSaver.from_conn_string(":memory:")
app = workflow.compile(checkpointer=memory)

# 5. 运行 LangGraph Agent

# 调用图：传入初始状态。对于消息列表，初始状态通常包含用户的第一条消息。
# config 用于配置执行，特别是 thread_id 用于状态持久化。
config = {"configurable": {"thread_id": "basic-agent-thread-1"}}

print("\n--- Starting Conversation 1 ---")
# 注意：invoke 方法返回最终状态
# 如果是循环，可能需要用户输入或外部事件来驱动
# 为了演示循环，我们将模拟多轮输入
inputs = [
    {"messages": [HumanMessage(content="你好，可以帮我写个简单的 Python 函数吗？")]},
    {"messages": [HumanMessage(content="函数名字叫 greet，接收一个名字参数，打印 'Hello, [名字]!'") ]},
    {"messages": [HumanMessage(content="写完了告诉我 说'完成' 就行。")]},
    {"messages": [HumanMessage(content="好的，谢谢你。再见！")]}
]

# 为了演示，我们手动模拟多轮调用，每次调用会从上一次结束的状态恢复并继续
# 在实际应用中，这可能发生在 Web 请求或消息队列处理中
current_state = None
for i, input_state in enumerate(inputs):
    print(f"\n--- User Input Round {i+1} ---")
    # 从上一次的状态继续，如果没有上一次状态，就使用当前 input 作为初始状态
    # 注意：invoke 如果有checkpointer和thread_id，会自动从最新状态恢复
    # 所以这里只需传入新增的消息即可（LangGraph会处理合并）
    # 第一次调用时，input_state 是 {"messages": [UserMessage(content="...") ]}
    # 后续调用时，只需要传入新增的 HumanMessage
    if i > 0:
         # 这里传入的 input 会被添加到上一次保存的状态中
         result = app.invoke(input_state, config=config)
    else:
         # 第一次调用，没有历史状态，从 input 开始
         result = app.invoke(input_state, config=config)

    current_state = result # 记录当前状态，虽然invoke with checkpointer and thread_id handles this
    print(f"--- Round {i+1} Result State ---")
    # 打印最终状态中的消息列表
    for msg in result['messages']:
        print(f"{type(msg).__name__}: {msg.content[:50]}...")

    # 检查是否到达 END 节点 (通过查看图的终点)
    # LangGraph invoke 方法返回的是最终状态，如果到达 END，执行就停止了
    # 我们可以根据循环是否结束来判断，或者检查 should_continue 返回值在外部控制
    # 在这个例子中，当 should_continue 返回 "end" 时，invoke 就会停止执行
    if should_continue(current_state) == "end":
        print("\n--- Conversation Ended ---")
        break

# 6. 演示状态持久化和恢复 (可选)
print("\n--- Demonstrating State Persistence ---")
# 模拟新的会话，但使用同一个 thread_id，将从上一次结束的状态开始
config_restore = {"configurable": {"thread_id": "basic-agent-thread-1"}}
print("Restoring state for thread 'basic-agent-thread-1'...")
# 传入新的输入，LangGraph会自动加载之前保存的状态，并将新输入添加到消息列表中
restored_result = app.invoke({"messages": [HumanMessage(content="请问刚才的 Python 函数写好了吗？")]}, config=config_restore)

print("\n--- Restored Conversation State ---")
for msg in restored_result['messages']:
    print(f"{type(msg).__name__}: {msg.content[:50]}...")

# 继续对话直到结束
print("\n--- Continue Restored Conversation ---")
final_input = {"messages": [HumanMessage(content="非常好，谢谢！这次真的再见了。")]}
final_result = app.invoke(final_input, config=config_restore)

print("\n--- Final State of Restored Conversation ---")
for msg in final_result['messages']:
    print(f"{type(msg).__name__}: {msg.content[:50]}...")

代码解释：

1. AgentState 定义：我们定义了一个 AgentState TypedDict，其中最重要的字段是 messages，用于存储 BaseMessage 对象的列表。 Annotated 和 operator.add 告诉 LangGraph 如何合并新返回的消息列表。

2. call_llm 节点：这是一个简单的函数，接收 AgentState，从中取出消息列表，调用 ChatOpenAI，并将 LLM 的回复作为一个 AIMessage 列表返回。 LangGraph 会将这个返回的 { "messages": [response] } 字典与当前状态合并，使用 operator.add 将新消息添加到 messages 列表中。

3. should_continue 节点：这是一个判断函数，接收 AgentState，检查最新的消息（通常是 LLM 回复）。根据消息内容，它返回 "continue" 或 "end"。

4. 图结构定义：

   • 创建 StateGraph(AgentState) 实例。

   • add_node("call_llm", call_llm)：将 call_llm 函数作为一个名为 "call_llm" 的节点添加到图中。

   • set_entry_point("call_llm")：设置 "call_llm" 为图的起始节点。

   • add_conditional_edges(...)：设置从 "call_llm" 节点出发的条件边。我们指定 should_continue 函数作为条件，并提供一个映射：如果函数返回 "continue"，转移到 "call_llm" 节点（形成循环）；如果返回 "end"，转移到 END 特殊节点（终止执行）。

5. 编译图： workflow.compile(checkpointer=memory) 将图结构编译成一个可执行的 LangGraph 应用。 checkpointer 配置开启了状态持久化。

6. 运行和状态持久化：我们使用 app.invoke() 调用编译后的图。传入 {"messages": [...]} 作为初始状态。 config={"configurable": {"thread_id": "..."}} 确保了状态可以按照指定的 thread_id 进行持久化和恢复。在循环中，每一次 invoke 如果使用相同的 config，都会先尝试从 checkpointer 加载该 thread_id 的最新状态，然后再处理本次输入。

这个基础示例展示了 LangGraph Agent 的基本构建模块：状态、节点、边、条件分支和循环，以及状态持久化。这是构建复杂 A2A 系统的基础。

第五章：高级 A2A 协作示例

基于 LangGraph 的核心概念，我们可以构建更复杂的 A2A 协作系统。本章将展示两种常见的 A2A 模式及其 LangGraph 实现。

5.1 示例 1: 研究员 Agent + 报告撰写 Agent 协作

场景描述:

一个用户需要一份关于某个主题的摘要报告。我们将有两个 Agent 协作完成任务：

1. 研究员 Agent (Researcher Agent)：负责根据主题进行网络搜索，收集相关信息。

2. 报告撰写 Agent (Report Writer Agent)：负责接收研究员 Agent 提供的资料，然后撰写一份摘要报告给用户。

协作流程:

1. 用户提交研究主题。

2. 研究员 Agent 接收主题，执行搜索。

3. 研究员 Agent 将搜索结果传递给报告撰写 Agent。

4. 报告撰写 Agent 接收搜索结果，撰写报告。

5. 报告撰写 Agent 将最终报告返回给用户，任务结束。

LangGraph 实现思路:

• 状态: 需要共享的状态包括用户请求、研究主题、搜索结果、最终报告等。

• 节点:

  • 一个节点用于初始化任务（接收主题）。

  • “研究员 Agent”节点：接收主题，调用搜索工具，更新状态中的搜索结果。

  • “报告撰写 Agent”节点：接收搜索结果，使用 LLM 生成报告，更新状态中的报告。

• 边:

  • 从开始到初始化节点。

  • 从初始化节点到“研究员 Agent”节点。

  • 从“研究员 Agent”节点到“报告撰写 Agent”节点。

  • 从“报告撰写 Agent”节点到 END 节点。

代码示例 (关键部分):

首先，我们需要定义工具（这里以模拟搜索工具为例）和 Agent 节点函数。

# 需要安装 duckduckgo-search 或其他搜索引擎工具
# pip install -U duckduckgo-search langchain-community

import operator
from typing import TypedDict, Annotated, List

from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langchain_community.tools import DuckDuckGoSearch

from langgraph.graph import StateGraph, END, START
from langgraph.checkpoint.sqlite import SqliteSaver

# 定义状态
class ResearchReportState(TypedDict):
    messages: Annotated[List[BaseMessage], operator.add]
    research_topic: str
    search_results: str
    final_report: str

# 初始化 LLM
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# 定义工具
@tool
def search_the_web(query: str) -> str:
    """Searches the web for information about the query."""
    print(f"--- Executing Search Tool for Query: '{query}' ---")
    search = DuckDuckGoSearch(max_results=5) # 限制搜索结果数量
    results = search.run(query)
    print("--- Search Complete ---")
    return results

# 定义 Agent 节点函数

# 研究员 Agent 节点
def researcher_agent(state: ResearchReportState) -> dict:
    """
    研究员 Agent 节点：从状态中获取研究主题，执行搜索，更新搜索结果状态。
    """
    print("--- Entering Researcher Agent ---")
    research_topic = state['research_topic']
    if not research_topic:
        print("Error: Research topic is missing.")
        return {"messages": [AIMessage(content="Error: Missing research topic.")]}

    try:
        # 使用 search_the_web 工具进行搜索
        # 注意：在 LangGraph 节点中直接调用工具函数，或封装到 Langchain Agent 中调用
        # 这里为了简化，直接调用工具函数。更复杂的Agent可以使用 Langchain AgentExecutor 节点
        search_query = f"关于 {research_topic} 的最新信息"
        search_results = search_the_web.invoke(search_query)

        print("--- Researcher Agent Finished ---")
        # 更新状态：添加一个消息说明搜索完成，并存储搜索结果
        return {
            "messages": [AIMessage(content=f"已获取关于 '{research_topic}' 的搜索结果。")],
            "search_results": search_results
        }
    except Exception as e:
        print(f"Researcher Agent Error: {e}")
        return {"messages": [AIMessage(content=f"研究员Agent执行失败：{e}")]}


# 报告撰写 Agent 节点
def report_writer_agent(state: ResearchReportState) -> dict:
    """
    报告撰写 Agent 节点：从状态中获取搜索结果，使用 LLM 撰写报告，更新最终报告状态。
    """
    print("--- Entering Report Writer Agent ---")
    research_results = state.get('search_results')
    research_topic = state.get('research_topic')

    if not research_results:
        print("Error: Search results missing.")
        return {"messages": [AIMessage(content="Error: Missing search results for report writing.")]}

    if not research_topic:
        print("Error: Research topic missing.")
        return {"messages": [AIMessage(content="Error: Missing research topic for report writing.")]}

    # 构建给 LLM 的提示，结合主题和搜索结果撰写报告
    prompt = f"""基于以下研究主题和搜索结果，撰写一份简洁的摘要报告（约200-300字）：

    研究主题：{research_topic}

    搜索结果：
    {research_results}

    请注意报告的流畅性和逻辑性。"""

    try:
        # 调用 LLM 撰写报告
        report_response = llm.invoke([HumanMessage(content=prompt)])
        final_report_content = report_response.content
        print("--- Report Writer Agent Finished ---")

        # 更新状态：添加一个消息说明报告撰写完成，并存储最终报告
        return {
            "messages": [AIMessage(content="报告撰写完成。")],
            "final_report": final_report_content
        }
    except Exception as e:
        print(f"Report Writer Agent Error: {e}")
        return {"messages": [AIMessage(content=f"报告撰写Agent执行失败：{e}")]}


# 初始化任务函数 (入口节点前的预处理)
def initialize_research_task(state: ResearchReportState) -> dict:
    """
    初始化任务，从用户输入中提取研究主题，并设置初始状态。
    """
    print("--- Initializing Research Task ---")
    # 假设用户输入消息的第一条包含了研究主题
    user_message_content = state['messages'][0].content
    # 简单地将整个用户输入作为研究主题
    research_topic = user_message_content
    print(f"Research Topic identified: '{research_topic}'")

    # 初始化状态，将研究主题保存到状态中
    return {
        "research_topic": research_topic,
        "messages": [AIMessage(content=f"已接收到您的研究主题：'{research_topic}'，正在进行研究...")]
    }


# 2. 构建图
workflow = StateGraph(ResearchReportState)

# 添加节点
workflow.add_node("initialize_task", initialize_research_task)
workflow.add_node("researcher", researcher_agent)
workflow.add_node("report_writer", report_writer_agent)

# 设置入口点
workflow.set_entry_point("initialize_task")

# 添加边：定义任务流
workflow.add_edge(START, "initialize_task") # 从开始到初始化
workflow.add_edge("initialize_task", "researcher") # 初始化后到研究员
workflow.add_edge("researcher", "report_writer") # 研究员完成后到报告撰写
workflow.add_edge("report_writer", END) # 报告撰写完成后结束

# 3. 编译图
# memory = SqliteSaver.from_conn_string(":memory:") # 如果需要持久化
# app = workflow.compile(checkpointer=memory)
app = workflow.compile() # 简单示例，不使用持久化

# 4. 运行图

print("\n--- Starting Research and Report Generation ---")
# 初始输入：用户消息包含研究主题
initial_input = {"messages": [HumanMessage(content="请研究一下 量子计算的最新进展，并写一份摘要。")]}
# config = {"configurable": {"thread_id": "research-report-1"}} # 如果使用持久化

final_state = app.invoke(initial_input) # 使用 invoke 运行图直到结束

# 5. 打印最终结果
print("\n--- Final Task State ---")
# 打印消息历史
print("Messages:")
for msg in final_state['messages']:
    print(f"- {type(msg).__name__}: {msg.content[:100]}...")

# 打印最终报告
print("\nFinal Report:")
print(final_state['final_report'])

代码解释:

1. ResearchReportState：定义了包含 messages, research_topic, search_results, final_report 的状态，用于在 Agent 之间传递信息。

2. search_the_web 工具：一个简单的工具封装，用于模拟网络搜索。在实际应用中，这会使用 Langchain 的 Tool 或 Runnable。

3. researcher_agent 节点：从状态中读取 research_topic，调用 search_the_web 获取结果，并将结果存回状态字典中。

4. report_writer_agent 节点：从状态中读取 search_results 和 research_topic，使用 LLM 根据这些信息生成报告，并将报告存回状态字典中。

5. initialize_research_task 节点：这是一个前置节点，负责从用户输入的初始消息中提取研究主题，并添加到状态中。

6. 图构建与边：使用 StateGraph 构建图，通过 add_node 添加三个 Agent 节点（initialize_task，researcher，report_writer）。使用 add_edge 定义了简单的顺序执行流程：初始化 -> 研究员 -> 报告撰写 -> 结束。

7. 运行：app.invoke(initial_input) 执行图。 LangGraph 会按照定义的边依次执行节点，每个节点执行后更新状态，直到到达 END 节点。

这个示例展示了如何通过 LangGraph 组织两个 Agent 按照预定的顺序协作完成一个任务，信息通过共享的 ResearchReportState 传递。

5.2 示例 2: 任务分解 Agent + 多个执行 Agent + 结果汇总 Agent

场景描述:

一个复杂的用户请求，需要分解成多个子任务，由不同的 Agent 并行或顺序执行，最后将结果汇总。我们将构建一个包含多个 Agent 的模式：

1. 任务分解 Agent (Task Decomposer Agent)：接收原始用户请求，将其分解为若干个独立的子任务列表。

2. 执行 Agent (Executor Agents)：负责执行特定类型的子任务（例如，搜索任务由搜索 Agent 执行，计算任务由计算 Agent 执行）。可能有多个不同类型的执行 Agent。

3. 结果汇总 Agent (Result Aggregator Agent)：接收所有子任务的执行结果，将它们整合、提炼，生成最终的统一答复给用户。

协作流程:

1. 用户提交复杂请求。

2. 任务分解 Agent 接收请求，分解成子任务列表，添加到状态。

3. 一个路由节点根据子任务类型，将任务分发给相应的执行 Agent。

4. 执行 Agent 完成任务，将结果添加到状态。

5. 所有子任务完成后，路由节点或另一个判断节点将流程转移到结果汇总 Agent。

6. 结果汇总 Agent 接收所有结果，生成最终答复，添加到状态。

7. 流程结束，返回最终答复。

LangGraph 实现思路:

• 状态: 需要共享的状态包括原始用户请求、子任务列表、每个子任务的状态（待执行、进行中、已完成、失败）、每个子任务的结果、最终汇总结果等。

• 节点:

  • “任务分解 Agent”节点。

  • 一个或多个“执行 Agent”节点（例如，“搜索执行 Agent”、“计算执行 Agent”）。

  • 一个条件路由节点：根据当前待执行的子任务类型，将执行流路由到对应的执行 Agent 节点。

  • 一个判断所有子任务是否完成的节点。

  • “结果汇总 Agent”节点。

• 边:

  • 从开始到任务分解 Agent。

  • 从任务分解 Agent 到路由节点。

  • 从路由节点到各个执行 Agent。

  • 从各个执行 Agent 回到判断所有子任务是否完成的节点。

  • 从判断节点有条件边：如果还有待执行任务，回到路由节点；如果所有任务完成，转移到结果汇总 Agent。

  • 从结果汇总 Agent 到 END 节点。

• 循环：这里会形成一个循环：路由 -> 执行 Agent -> 判断完成 -> (未完成) 回到路由。

代码示例 (关键部分):

由于这个示例涉及多个 Agent、复杂的路由和状态管理，我们将重点展示其核心结构和关键节点函数的设计。完整的代码会比较长。

import operator
from typing import TypedDict, Annotated, List, Dict, Any

from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

from langgraph.graph import StateGraph, END, START

# 假设我们有一些工具
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

@tool
def search_tool(query: str) -> str:
    """Searches the web for information about the query."""
    print(f"--- Search Tool: Searching for '{query}' ---")
    # 真实场景会使用实际的搜索 API
    return f"搜索结果 for '{query}': 模拟搜索结果..."

@tool
def calculator(expression: str) -> str:
    """Evaluates a mathematical expression."""
    print(f"--- Calculator Tool: Evaluating '{expression}' ---")
    try:
        return str(eval(expression)) # 模拟计算
    except Exception as e:
        return f"计算错误: {e}"

# 定义子任务的结构
class SubTask(TypedDict):
    id: str # 任务唯一ID
    type: str # 任务类型 (e.g., "search", "calculate", "write", "plan")
    description: str # 任务描述
    status: str # 任务状态 (e.g., "pending", "running", "completed", "failed")
    result: Any # 任务结果 (完成后填充)
    assigned_agent: str # 分派给哪个 Agent 执行

# 定义 AgentState
class TaskCollaborationState(TypedDict):
    messages: Annotated[List[BaseMessage], operator.add]
    original_request: str
    sub_tasks: Annotated[List[SubTask], operator.add] # 子任务列表
    # 可以添加其他状态，例如：
    # final_answer: str

# 定义 Agent 节点函数

# 任务分解 Agent 节点
def task_decomposer_agent(state: TaskCollaborationState) -> dict:
    """
    任务分解 Agent：接收原始请求，分解成子任务列表。
    使用 LLM 来执行分解逻辑。
    """
    print("--- Entering Task Decomposer Agent ---")
    original_request = state['messages'][0].content # 假设原始请求在第一条消息中
    print(f"Original Request: '{original_request}'")

    # 使用 LLM 进行任务分解，让 LLM 输出子任务列表的 JSON 格式
    # 实际中需要设计精巧的提示词
    prompt = f"""请将以下用户请求分解为一系列独立的、可执行的子任务列表，每个子任务包含 type（可选类型：search, calculate, write, generic），description。请以 JSON 格式输出列表。

    用户请求：{original_request}

    JSON 格式示例：
    [
      {{"id": "task_1", "type": "search", "description": "关于xxx的最新研究"}},
      {{"id": "task_2", "type": "calculate", "description": "计算 100 * 1.05"}},
      ...
    ]
    """
    task_list_str = llm.invoke([HumanMessage(content=prompt)]).content

    # 解析 LLM 输出的 JSON 字符串为 Python 列表
    # 需要处理解析错误
    try:
        import json
        task_list_raw = json.loads(task_list_str.strip())
        sub_tasks: List[SubTask] = []
        for task_data in task_list_raw:
            # 补充状态信息
            task_data['status'] = 'pending'
            task_data['result'] = None
            task_data['assigned_agent'] = None
            sub_tasks.append(SubTask(task_data)) # 确保是 SubTask 类型
        print(f"Decomposed into {len(sub_tasks)} sub-tasks.")
        # 返回更新状态
        return {
            "original_request": original_request,
            "sub_tasks": sub_tasks,
            "messages": [AIMessage(content=f"已将请求分解为 {len(sub_tasks)} 个子任务。")]
            }
    except Exception as e:
        print(f"Error decomposing task or parsing JSON: {e}")
        return {
            "messages": [AIMessage(content=f"任务分解失败或结果解析错误：{e}")],
            "sub_tasks": [] # 清空任务列表或返回错误状态
            }

# 条件路由节点：根据待执行任务类型决定下一个 Agent
def route_task(state: TaskCollaborationState) -> str:
    """
    查找第一个状态为 'pending' 的子任务，根据其类型决定路由到哪个执行 Agent。
    如果没有 pending 任务，表示所有任务已处理，路由到汇总 Agent。
    """
    print("--- Routing Task ---")
    sub_tasks = state.get('sub_tasks', [])
    for task in sub_tasks:
        if task['status'] == 'pending':
            print(f"Routing task '{task['id']}' (Type: {task['type']})")
            # 根据任务类型返回对应的 Agent 节点名称
            if task['type'] == 'search':
                return "search_executor"
            elif task['type'] == 'calculate':
                return "calculate_executor"
            # 添加更多类型...
            elif task['type'] == 'generic': # 如果没有特定 Agent，可以路由到通用 Agent
                 return "generic_executor"
            else:
                 print(f"Warning: Unknown task type '{task['type']}', routing to generic.")
                 return "generic_executor" # 未知类型也路由到通用 Agent

    # 如果没有 pending 任务，则所有任务已完成，进入汇总阶段
    print("All sub-tasks processed. Routing to aggregator.")
    return "aggregate_results"

# 判断所有子任务是否完成的节点
# 这个节点在流程中被执行到，检查状态并决定是否继续路由或进入汇总
def check_all_tasks_completed(state: TaskCollaborationState) -> str:
     """
     检查所有子任务的状态，如果全部完成则返回 'completed'，否则返回 'pending'。
     这个函数通常作为条件边从执行 Agent 指向，并决定是回到路由还是进入汇总。
     """
     print("--- Checking All Tasks Completion ---")
     sub_tasks = state.get('sub_tasks', [])
     all_completed = True
     for task in sub_tasks:
          if task['status'] in ['pending', 'running']:
               all_completed = False
               break # 发现有未完成任务，直接退出
     if all_completed:
          print("All sub-tasks are completed.")
          return "completed"
     else:
          print("Some tasks are still pending/running.")
          # 注意：这种设计中，如果任务还在进行中，需要在流程上确保它会回到可以检查任务状态的地方或等待
          # 简单的同步模型可以回到路由，由路由查找下一个 pending 任务（如果还有）
          # 更复杂的异步模型可能需要不同的处理
          return "pending" # 在同步流程中，'pending' 表示需要继续查找并执行下一个任务


# 执行 Agent 示例：搜索执行 Agent
def search_executor_agent(state: TaskCollaborationState) -> dict:
    """
    搜索执行 Agent：查找第一个 pending 的搜索任务并执行。
    """
    print("--- Entering Search Executor Agent ---")
    sub_tasks = state.get('sub_tasks', [])
    # 找到第一个 pending 的 search 任务
    task_to_execute = None
    for task in sub_tasks:
        if task['type'] == 'search' and task['status'] == 'pending':
            task_to_execute = task
            break

    if task_to_execute:
        print(f"Executing search task: {task_to_execute['description']}")
        # 更新任务状态为 running (可选，取决于是否需要精细状态)
        # task_to_execute['status'] = 'running' # 直接修改列表中的字典会影响原始状态

        try:
            # 执行搜索工具
            search_result = search_tool.invoke(task_to_execute['description'])
            # 更新任务状态为 completed 并保存结果
            task_to_execute['status'] = 'completed'
            task_to_execute['result'] = search_result
            task_to_execute['assigned_agent'] = "search_executor"
            print(f"Search task '{task_to_execute['id']}' completed.")
            # 返回更新后的子任务列表以反映状态变化
            return {"sub_tasks": sub_tasks} # operator.add for lists will replace the list if no key specified? Needs careful handling.

        except Exception as e:
            print(f"Search task '{task_to_execute['id']}' failed: {e}")
            task_to_execute['status'] = 'failed'
            task_to_execute['result'] = f"Execution failed: {e}"
            task_to_execute['assigned_agent'] = "search_executor"
            return {"sub_tasks": sub_tasks} # 返回更新后的列表

    else:
        print("No pending search tasks found.")
        # 如果没有找到 pending 任务，这个节点可能不应该被路由到，
        # 或者需要有机制处理这种情况（例如返回一个标识，在 check_all_tasks_completed 中处理）
        # 为了演示，这里简单打印，不修改状态
        return {"sub_tasks": sub_tasks} # 返回原始列表

# 执行 Agent 示例：计算执行 Agent (与搜索执行类似)
def calculate_executor_agent(state: TaskCollaborationState) -> dict:
    """
    计算执行 Agent：查找第一个 pending 的 calculate 任务并执行。
    """
    print("--- Entering Calculate Executor Agent ---")
    sub_tasks = state.get('sub_tasks', [])
    task_to_execute = None
    for task in sub_tasks:
        if task['type'] == 'calculate' and task['status'] == 'pending':
            task_to_execute = task
            break

    if task_to_execute:
        print(f"Executing calculate task: {task_to_execute['description']}")
        try:
            calc_result = calculator.invoke(task_to_execute['description'])
            task_to_execute['status'] = 'completed'
            task_to_execute['result'] = calc_result
            task_to_execute['assigned_agent'] = "calculate_executor"
            print(f"Calculate task '{task_to_execute['id']}' completed.")
            return {"sub_tasks": sub_tasks} # 返回更新后的列表
        except Exception as e:
            print(f"Calculate task '{task_to_execute['id']}' failed: {e}")
            task_to_execute['status'] = 'failed'
            task_to_execute['result'] = f"Execution failed: {e}"
            task_to_execute['assigned_agent'] = "calculate_executor"
            return {"sub_tasks": sub_tasks} # 返回更新后的列表
    else:
        print("No pending calculate tasks found.")
        return {"sub_tasks": sub_tasks} # 返回原始列表

# 结果汇总 Agent 节点
def result_aggregator_agent(state: TaskCollaborationState) -> dict:
    """
    结果汇总 Agent：接收所有已完成的子任务结果，使用 LLM 生成最终答复。
    """
    print("--- Entering Result Aggregator Agent ---")
    original_request = state.get('original_request')
    sub_tasks = state.get('sub_tasks', [])

    if not sub_tasks:
        final_answer = "未能生成任何子任务，请检查请求。"
        print("No sub-tasks found for aggregation.")
    else:
        completed_results = [
            f"Task ID: {task['id']}\nType: {task['type']}\nDescription: {task['description']}\nResult: {task['result']}\nStatus: {task['status']}"
            for task in sub_tasks if task['status'] == 'completed'
        ]
        failed_tasks = [
             f"Task ID: {task['id']}, Description: {task['description']}, Status: {task['status']}, Result: {task['result']}"
             for task in sub_tasks if task['status'] != 'completed'
        ]

        if not completed_results:
            final_answer = f"未成功完成任何子任务。失败任务：{failed_tasks}"
            print("No completed tasks for aggregation.")
        else:
            # 使用 LLM 汇总结果
            prompt = f"""原始用户请求是：{original_request}

            以下是各子任务的执行结果：
            {chr(10).join(completed_results)}

            已失败或未执行的任务：
            {'无失败任务' if not failed_tasks else chr(10).join(failed_tasks)}

            请根据原始请求和已完成任务的结果，以及失败任务的说明，生成一个最终的、简洁、完整且易于理解的答复。"""

            try:
                final_answer_response = llm.invoke([HumanMessage(content=prompt)])
                final_answer = final_answer_response.content
                print("--- Result Aggregation Complete ---")
            except Exception as e:
                 final_answer = f"结果汇总失败：{e}"
                 print(f"Result Aggregator Agent Error: {e}")

    # 更新状态，将最终答复添加到消息列表
    return {
        "messages": [AIMessage(content=final_answer)],
        # 可以将最终答复单独存储在状态中
        # "final_answer": final_answer
    }


# 2. 构建图结构

workflow = StateGraph(TaskCollaborationState)

# 添加节点
workflow.add_node("decomposer", task_decomposer_agent)
workflow.add_node("route_task", route_task)
workflow.add_node("search_executor", search_executor_agent)
workflow.add_node("calculate_executor", calculate_executor_agent)
workflow.add_node("aggregate_results", result_aggregator_agent)
# workflow.add_node("check_completion", check_all_tasks_completed) # 可以作为单独节点，也可以合并到 route_task 或作为边条件


# 设置入口点
workflow.set_entry_point("decomposer") # 从任务分解开始

# 添加边
workflow.add_edge(START, "decomposer") # 开始 -> 分解任务
workflow.add_edge("decomposer", "route_task") # 分解任务 -> 路由


# 从路由节点出发到不同的执行 Agent (条件边在此处实现)
# 注意：这里的 route_task 函数返回的是下一个节点名称
workflow.add_conditional_edges(
    "route_task",         # 起始节点
    route_task,           # 条件判断函数（其实就是 route_task 节点本身的功能）
    {                     # 映射
        "search_executor": "search_executor",
        "calculate_executor": "calculate_executor",
        # 可以添加其他类型...
        "generic_executor": "generic_executor", # 如果设计了通用执行器
        "aggregate_results": "aggregate_results" # 如果 route_task 判断没有 pending 任务，直接到汇总
    }
)

# 从执行 Agent 返回检查任务完成状态，然后回到路由节点继续处理其他任务（形成循环）
workflow.add_edge("search_executor", "route_task")     # 搜索完成后回到路由
workflow.add_edge("calculate_executor", "route_task")   # 计算完成后回到路由
# ... 其他执行 Agent 也指向 route_task

# 从汇总节点到 END
workflow.add_edge("aggregate_results", END)

# 3. 编译图
app = workflow.compile()

# 4. 运行图

print("\n--- Starting Task Decomposition and Collaboration ---")
# 初始输入：用户消息包含复杂请求
initial_request = "请帮我研究量子计算的最新进展，计算 123 + 456 的结果，并写一份关于这两个内容的简要报告。"
initial_input = {"messages": [HumanMessage(content=initial_request)]}

final_state = app.invoke(initial_input)

# 5. 打印最终结果
print("\n--- Final Task Collaboration State ---")
print("Messages:")
for msg in final_state['messages']:
    print(f"- {type(msg).__name__}: {msg.content[:100]}...")

print("\nSub-tasks Status:")
for task in final_state.get('sub_tasks', []):
     print(f"- [{task['id']}] Type: {task['type']}, Status: {task['status']}, Assigned: {task['assigned_agent']}, Result: {task['result']}")

print("\nFinal Answer:")
# 最终答复通常是状态中最后一条 AIMessage 或存储在特定字段中
# 在这个例子中，aggregator 将最终答复添加到了 messages 列表
print(final_state['messages'][-1].content)

代码解释:

1. SubTask 和 TaskCollaborationState：定义了 SubTask 结构来描述每个子任务，以及包含原始请求和子任务列表的 TaskCollaborationState。注意到 sub_tasks 使用了 operator.add，但在节点函数中直接修改列表的元素并返回整个列表来更新状态可能需要更精细的处理，或者使用其他状态合并策略。一个更安全的方法可能是返回更新后的子任务列表的副本或者使用自定义的合并函数。在 LangGraph 的最新版本中，直接修改 mutable 对象（如列表、字典）并在节点函数中返回包含这些对象的字典通常会按引用传递并更新，但这在理解上可能比不可变数据结构的合并更复杂。这里示例直接修改并返回列表，假定其能正确反映状态变化。

2. task_decomposer_agent 节点：接收用户请求，调用 LLM 将其分解为结构化的子任务列表，并初始化子任务的状态（pending）。

3. route_task 节点：这是实现任务分发的关键。它遍历 sub_tasks 列表，找到第一个状态为 pending 的任务，并根据其 type 返回对应的执行 Agent 节点的命名字符串。如果找不到 pending 任务，则返回 "aggregate_results"。这个函数既是节点本身的功能，也作为从该节点出发的条件边的判断函数。

4. 执行 Agent 节点 (search_executor_agent, calculate_executor_agent)：这些节点负责执行特定类型的子任务。它们会找到一个待执行的相应类型任务，调用相应的工具，更新任务的状态和结果，并返回修改后的 sub_tasks 列表。

5. check_all_tasks_completed 节点 (可选)：一个辅助判断节点，用于检查所有子任务是否都已完成。虽然在上面的简化图中没有作为单独节点使用（逻辑集成到 route_task 或边条件中），但它可以作为一个清晰的流程步骤。

6. result_aggregator_agent 节点：在所有子任务完成后执行。它收集所有已完成任务的结果，调用 LLM 整合信息，生成最终给用户的答复。

7. 图构建与边：Graph 连接了分解器、路由和执行 Agent。 核心是条件边从 route_task 到不同的执行 Agent，以及从执行 Agent 回到 route_task，形成处理子任务的循环。 一旦 route_task 判断所有子任务完成，流程通过条件边转移到 aggregate_results 节点。

这个复杂示例展示了 LangGraph 如何通过状态管理和条件边实现动态的任务分发和协作循环，多个 Agent 可以根据任务类型和系统状态被动态调用。

第六章：A2A 架构模式与 LangGraph 实现

多 Agent 系统可以采用多种不同的架构模式，以适应不同的需求和复杂性。 LangGraph 的灵活性使其能够支持多种这些模式的实现。

6.1 常见的 A2A 架构模式

• 顺序协作 (Sequential Collaboration)： 最简单的模式，Agent 按照预定的顺序依次执行。一个 Agent 的输出作为下一个 Agent 的输入（通过共享状态）。示例 1 (研究员+报告撰写) 就属于这种模式。

  <svg width="300" height="100" xmlns="http://www.w3.org/2000/svg">
    <rect x="10" y="30" width="80" height="40" fill="#ADD8E6" stroke="#000"/>
    <text x="50" y="55" text-anchor="middle">Agent A</text>
    <rect x="110" y="30" width="80" height="40" fill="#ADD8E6" stroke="#000"/>
    <text x="150" y="55" text-anchor="middle">Agent B</text>
    <rect x="210" y="30" width="80" height="40" fill="#ADD8E6" stroke="#000"/>
    <text x="250" y="55" text-anchor="middle">Agent C</text>
    <line x1="90" y1="50" x2="110" y2="50" stroke="#000" marker-end="url(#arrow)"/>
    <line x1="190" y1="50" x2="210" y2="50" stroke="#000" marker-end="url(#arrow)"/>
    <defs>
        <marker id="arrow" markerWidth="10" markerHeight="10" refX="9" refY="5" orient="auto">
            <path d="M0,0 L10,5 L0,10 Z" fill="#000" />
        </marker>
    </defs>
  </svg>

  LangGraph 实现: 使用 add_edge(node_a, node_b) 按顺序连接节点。

• 层级式 (Hierarchical)： 有一个或多个“监督 Agent”负责整体协调、任务分解和结果汇总。底层是执行具体任务的“工作 Agent”。监督 Agent 将任务分发给工作 Agent，并接收其结果报告。示例 2 (任务分解+执行+汇总) 就属于层级式的一种变体。

  <svg width="300" height="200" xmlns="http://www.w3.org/2000/svg">
    <rect x="110" y="10" width="80" height="40" fill="#ADD8E6" stroke="#000"/>
    <text x="150" y="35" text-anchor="middle">Supervisor Agent</text>
  
    <rect x="10" y="100" width="80" height="40" fill="#90EE90" stroke="#000"/>
    <text x="50" y="125" text-anchor="middle">Worker A</text>
    <rect x="110" y="100" width="80" height="40" fill="#90EE90" stroke="#000"/>
    <text x="150" y="125" text-anchor="middle">Worker B</text>
    <rect x="210" y="100" width="80" height="40" fill="#90EE90" stroke="#000"/>
    <text x="250" y="125" text-anchor="middle">Worker C</text>
  
    <line x1="150" y1="50" x2="50" y2="100" stroke="#000" marker-end="url(#arrow)"/>
    <line x1="150" y1="50" x2="150" y2="100" stroke="#000" marker-end="url(#arrow)"/>
    <line x1="150" y1="50" x2="250" y2="100" stroke="#000" marker-end="url(#arrow)"/>
  
    <line x1="50" y1="140" x2="150" y2="50" stroke="#000" marker-end="url(#arrow)" transform="rotate(180 100 120)"/>
    <line x1="150" y1="140" x2="150" y2="50" stroke="#000" marker-end="url(#arrow)" transform="rotate(180 150 95)"/>
    <line x1="250" y1="140" x2="150" y2="50" stroke="#000" marker-end="url(#arrow)" transform="rotate(180 200 120)"/>
  
    <defs>
        <marker id="arrow" markerWidth="10" markerHeight="10" refX="9" refY="5" orient="auto">
            <path d="M0,0 L10,5 L0,10 Z" fill="#000" />
        </marker>
    </defs>
  </svg>

  LangGraph 实现: 使用一个节点作为 Supervisor，该节点（或它指向的条件边）根据任务类型路由到不同的 Worker 节点。Worker 完成后，流程回到 Supervisor 节点（或一个检查节点），进行结果处理或决定下一个任务。

• 分布式/网络式 (Distributed/Network)： 没有严格的层级关系，Agent 之间可以自由通信并动态决定下一个将要接收控制权的 Agent。Agent 可能会将任务“移交 (handoffs)”给最合适的 Agent。

  <svg width="300" height="200" xmlns="http://www.w3.org/2000/svg">
   <rect x="10" y="10" width="80" height="40" fill="#ADD8E6" stroke="#000"/>
   <text x="50" y="35" text-anchor="middle">Agent A</text>
   <rect x="210" y="10" width="80" height="40" fill="#ADD8E6" stroke="#000"/>
   <text x="250" y="35" text-anchor="middle">Agent B</text>
   <rect x="110" y="100" width="80" height="40" fill="#ADD8E6" stroke="#000"/>
   <text x="150" y="125" text-anchor="middle">Agent C</text>
   <rect x="10" y="150" width="80" height="40" fill="#ADD8E6" stroke="#000"/>
   <text x="50" y="175" text-anchor="middle">Agent D</text>
   <rect x="210" y="150" width="80" height="40" fill="#ADD8E6" stroke="#000"/>
   <text x="250" y="175" text-anchor="middle">Agent E</text>
  
   <line x1="90" y1="30" x2="210" y2="30" stroke="#000" marker-end="url(#arrow)"/>
   <line x1="50" y1="50" x2="110" y2="100" stroke="#000" marker-end="url(#arrow)"/>
   <line x1="250" y1="50" x2="190" y2="100" stroke="#000" marker-end="url(#arrow)"/>
   <line x1="150" y1="140" x2="50" y2="150" stroke="#000" marker-end="url(#arrow)"/>
   <line x1="150" y1="140" x2="250" y2="150" stroke="#000" marker-end="url(#arrow)"/>
   <line x1="90" y1="170" x2="210" y2="170" stroke="#000" marker-end="url(#arrow)"/>
  
    <defs>
       <marker id="arrow" markerWidth="10" markerHeight="10" refX="9" refY="5" orient="auto">
           <path d="M0,0 L10,5 L0,10 Z" fill="#000" />
       </marker>
   </defs>
  </svg>

  LangGraph 实现: 每个 Agent 节点内部的逻辑（通常通过 LLM 的推理决定）决定返回哪个下一个 Agent 节点的名称，然后通过条件边转移。可以使用 LangGraph 的 Command(goto=...) 机制实现更灵活的控制流移交³⁵。

• 市场式 (Market-based)： Agent 通过某种形式的“市场”（如拍卖、投标）来获取任务或资源。这通常需要额外的协调机制和协议。 LangGraph 实现: 可以在 LangGraph 内部模拟市场机制，例如一个节点负责“发布任务”，其他 Agent 节点“投标”，然后一个节点“分配任务”。但这会比前面几种模式复杂得多，可能需要结合外部服务。

6.2 LangGraph 对架构模式的支持

LangGraph 通过其核心组件提供了对这些架构模式的强大支持：

• 节点 (add_node)：可以代表任何类型的 Agent（监督、工作、专业执行器等）。

• 普通边 (add_edge)：实现顺序执行流。

• 条件边 (add_conditional_edges, 条件判断函数)：实现层级式和分布式架构中的动态路由和任务分发。

• 状态 (StateGraph, TypedDict)：提供 Agent 间共享信息和同步状态的通道。

• 循环：支持 Agent 之间的多轮交互和迭代过程（例如 ReAct 循环、任务分解/执行循环）。

• 子图 (SubGraph)：可以将一个复杂的 Agent 或一个 Agent 团队封装成一个子图节点，然后在主图中使用。这支持构建更复杂的层级结构。

• Handoffs (Command(goto=...))：在网络式和分布式架构中，Agent 可以直接指定下一个接收控制权的 Agent 名称，实现流畅的移交³⁵。

通过组合这些功能，开发者可以灵活地设计和实现各种 A2A 架构模式。

第七章：代理间通信：消息传递与状态共享

在 LangGraph 构建的 A2A 系统中，Agent 之间的通信主要通过两种方式实现：共享状态的更新和消息列表的传递。这是实现 Agent 协作的关键。

7.1 基于共享状态的通信

如前所述，LangGraph 中的所有 Agent 节点都共享一个 AgentState 对象。Agent 通过修改和访问这个共享状态来进行信息交换。

• 数据传递: 一个 Agent 完成其任务后，可以将结果数据存储在 AgentState 的相应字段中。下一个需要这些数据的 Agent 节点可以从状态中读取这些数据进行处理。

  # Agent A 完成任务，将结果存入状态
  def agent_a_node(state: YourAgentState) -> dict:
      result_a = process_data(state['input_data'])
      return {"result_of_A": result_a}
  
  # Agent B 从状态中读取 Agent A 的结果
  def agent_b_node(state: YourAgentState) -> dict:
      data_from_a = state.get('result_of_A')
      if data_from_a:
          final_output = analyze_result(data_from_a)
          return {"final_output": final_output}
      else:
           # 处理 Agent A 结果缺失的情况
           return {}
  

• 控制信号/标志: Agent 可以在状态中设置标志或状态变量，以指示其完成状态、遇到的问题或下一步建议。其他 Agent 或路由节点可以根据这些标志来改变流程走向。

  class TaskState(TypedDict):
      task_id: str
      status: str # pending, completed, failed
      assigned_agent: str
      result: Any
  
  class AgentState(TypedDict):
      tasks: Annotated[List[TaskState], lambda existing, new: ...] # 自定义合并逻辑以更新任务列表
      should_proceed_to_next_step: bool # Agent A 设置此标志
  
  def agent_a_node(state: AgentState) -> dict:
     # ... 完成任务逻辑 ...
     return {"should_proceed_to_next_step": True}
  
  # 条件边或判断节点可以检查 state['should_proceed_to_next_step'] 来决定流程
  def check_flag(state: AgentState) -> str:
      if state.get('should_proceed_to_next_step'):
          return "next_step"
      else:
          return "wait"

• 复杂数据结构: 共享状态可以包含复杂的 Python 对象，如列表、字典、甚至是自定义类的实例，只要它们可以被序列化以便持久化（如果开启了状态持久化）。对于列表和字典等可变对象，更新状态时需要注意合并策略（如使用 operator.add 进行列表拼接，或自定义函数进行更复杂的字典更新）。

7.2 基于消息列表的通信 (MessagesState)

Langchain 的 BaseMessage 列表是 LLM 交互的标准格式。 LangGraph 通常会将消息列表作为 AgentState 的核心部分 (MessagesState)。 Agent 之间的许多通信都是通过向这个共享的消息列表中添加消息来实现的。

• ReAct 模式通信: 在基于 ReAct 模式构建的 Agent 中，LLM 的思考过程 (scratchpad)、工具调用 (ToolMessage) 和工具输出 (ToolMessage) 都被格式化为 BaseMessage 类型并添加到消息列表中。 Agent 下一次运行时，LLM 可以通过完整的消息历史来回忆之前的步骤、思考过程和工具执行结果。这是 Agent 内部推理的重要通信方式。

• Agent 间对话: 在一些多 Agent 模式中，Agent 之间可以通过模拟对话的方式进行通信。一个 Agent 生成一条消息（例如 AIMessage 或自定义消息类型），包含它要传达的信息或指令，并将其添加到共享消息列表中。下一个 Agent 则可以读取列表中的新消息，理解其内容并做出反应。为了区分消息来源，可以使用消息的 name 字段或在消息内容中明确标记发送 Agent 的身份。

  from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
  
  # 在 Agent 节点中发送消息给另一个 Agent
  def agent_a_node(state: MessagesState) -> dict:
      # Agent A 准备发送一个内部消息给 Agent B
      internal_message_to_b = AIMessage(
          content="请帮我验证一下这个计算结果：100+200 是不是 300",
          name="agent_a" # 可选：标记发送者身份
          )
      # 将消息添加到共享列表
      return {"messages": [internal_message_to_b]}
  
  # Agent B 节点读取消息列表并查找需要自己处理的消息
  def agent_b_node(state: MessagesState) -> dict:
      messages = state['messages']
      # 遍历消息，查找发送给自己的指令或信息
      instruction_from_a = None
      for msg in reversed(messages): # 从最新消息开始查找
           if isinstance(msg, AIMessage) and msg.name == "agent_a" and "验证计算结果" in msg.content:
                instruction_from_a = msg.content
                break # 找到了需要处理的消息
  
      if instruction_from_a:
          print("Agent B received instruction from Agent A:", instruction_from_a)
          # 执行验证逻辑...
          verification_result = "验证结果是正确的。"
          # 将结果作为消息添加到列表，发给 Agent A 或用户
          response_message = AIMessage(content=f"Agent B 回复 Agent A: 验证完成，{verification_result}", name="agent_b")
          return {"messages": [response_message]}
      else:
          print("Agent B found no specific instruction from Agent A.")
          return {} # 没有需要处理的消息，不修改状态
  

  这种消息传递方式的好处是所有通信都可以记录在统一的消息历史中，便于 LangSmith 等工具进行追踪和调试。它模拟了 Agent 之间通过“聊天记录”进行协作³⁵。

7.3 实现 Agent 间通信的技巧

• 清晰的状态定义: 精心设计 AgentState TypedDict 的结构，确保所有 Agent 需要共享的信息都有明确的字段。使用 Annotated 配合合适的合并策略（operator.add、operator.setitem 或自定义函数）来控制状态更新行为。

• 规范消息格式: 如果通过消息列表通信，考虑定义一套内部消息格式或约定，例如使用特定的前缀、标签或消息的 name 字段来标识消息的类型、发送者和接收者。

• 利用条件边路由: Agent 在完成任务或需要将控制权移交给另一个 Agent 时，可以在返回状态时包含一个指示下一个步骤的信息（例如，一个状态字段或一个特殊的 Command 对象），然后利用条件边根据这个信息路由到下一个 Agent 节点。

• 自定义状态合并 (StateGraph(..., reducer=...)): 对于复杂的、多 Agent 同时修改状态的情况（尽管 LangGraph 的同步执行模型通常一次只有一个节点在运行），或者需要复杂的逻辑（如冲突解决、数据聚合）来合并状态更新时，可以自定义状态的 reducer 函数。

• Handoffs (Command 对象): LangGraph 的 Command 对象提供了一种显式的控制流移交机制。 Agent 节点可以返回 Command(goto="next_agent_name", update={"state_key": value}) 来指定下一个节点并同时更新状态³⁵。这比仅仅通过条件边判断更加直接和灵活。

  from langgraph.types import Command
  
  def agent_a_handoffs(state: AgentState) -> Command:
      # Agent A 完成任务后，决定将控制权移交给 Agent B
      print("--- Agent A finishing, handing off to Agent B ---")
      return Command(
          goto="agent_b", # 指定下一个节点名称
          update={"status_a": "completed"} # 同时可以更新状态
          # grap=Command.PARENT 或 Command.SUBGRAPH if using subgraphs
      )
  

  在图构建时，边需要能够响应这种 Command 返回。LangGraph 通常会自动处理以 Command 对象作为返回值的节点的边。

通过有效利用共享状态（特别是消息列表）和条件边（以及 Handoffs），您可以设计出 Agent 之间流畅、高效且可追踪的通信和协作逻辑。

第八章：调试和测试 A2A 系统

多 Agent 系统的复杂性使得调试和测试变得尤为重要，也更具挑战性。 Agent 之间的交互、动态的控制流和状态变化都增加了排查问题的难度。 LangGraph 和 Langchain 生态系统提供了一些工具和技术来帮助进行调试和测试。

8.1 使用 Print 语句和日志

最基础的调试方法是在 Agent 节点函数中加入 print 语句或使用 Python 内置的 logging 模块输出关键信息，例如：

• Agent 节点的开始和结束。

• 从状态中读取的关键数据。

• 调用外部工具的输入和输出。

• 条件判断函数的返回结果。

• 状态更新前后的内容。

# 在节点函数中
def my_agent_node(state: AgentState) -> dict:
    print(f"--- Entering my_agent_node. Current state keys: {state.keys()} ---")
    # ... 节点逻辑 ...
    print(f"--- Exiting my_agent_node. Returning update: {update_dict} ---")
    return update_dict

这种方法简单直接，适用于快速验证小规模的 Agent 工作流。但对于复杂的图结构、循环和多轮交互，仅仅依靠 print 输出信息可能会非常零散，难以追踪完整的执行路径和状态演变。

8.2 使用 LangSmith 进行可视化追踪

如前文所述，强烈推荐使用 LangSmith。将 LangSmith 环境变量配置好后，LangGraph 的每次 invoke 调用都会在 LangSmith UI 中生成一个详细的追踪记录（Trace）。

LangSmith Trace 提供：

• 可视化的执行图: 展示了 Agent 系统的执行流程图，每个节点、每条边的转移都清晰可见。

• 节点详细信息: 点击图中的节点，可以查看该节点的输入状态、输出、执行时间、调用的 LLM 的输入和输出、工具调用的详细信息等。

• 状态变化: 可以查看进入和离开每个节点时 AgentState 的具体内容，帮助理解状态是如何在 Agent 之间传递和更新的。

• 错误信息: 如果某个节点执行失败，LangSmith 会标记错误，并提供详细的错误堆栈信息。

• LLM 调用和工具调用细节: Langchain 提供的 LangSmith 集成会展示 LLM 调用和工具调用的完整请求和响应，便于排查 LLM 理解提示词或工具调用参数的问题。

通过 LangSmith，您可以直观地看到数据流和控制流在 Agent 之间如何流动，快速定位问题发生的节点和原因。这对于调试复杂的 Agent 协作和状态管理问题至关重要。

8.3 单元测试和集成测试

• 单元测试: 对每个 Agent 节点函数、条件判断函数、自定义状态合并函数等核心组件进行单元测试。隔离测试可以确保每个 Agent 或功能模块按照预期工作。

  • 测试节点函数: 模拟输入状态，调用节点函数，断言返回的状态更新字典是否正确，以及函数是否执行了预期的副作用（如工具调用）。

  • 测试条件判断函数: 模拟不同状态作为输入，断言函数返回的下一个节点名称是否正确。

  # 单元测试示例 (使用 pytest)
  def test_researcher_agent_with_topic():
      initial_state = ResearchReportState(messages=[], research_topic="量子计算", search_results="", final_report="")
      # 模拟 search_the_web 工具返回固定结果，避免真实网络调用
      # 可以使用 mock 库实现
      # 例如： mocker.patch('your_module.search_the_web', return_value="mocked search results")
  
      # 调用节点函数
      update = researcher_agent(initial_state)
  
      # 断言返回的状态更新是否符合预期
      assert "messages" in update
      assert isinstance(update["messages"][0], AIMessage)
      assert "research_results" in update
      assert update["search_results"] is not None # 或者更具体的断言
  
  
  def test_should_continue_end_condition():
      # 模拟一个包含结束关键词的 AIMessage
      state_with_end_message = AgentState(messages=[HumanMessage(content="话题"), AIMessage(content="好的，我们下次再聊。再见！")])
      # 调用条件判断函数
      result = should_continue(state_with_end_message)
      # 断言结果
      assert result == "end"

• 集成测试: 测试整个 LangGraph 工作流。模拟初始输入，调用 app.invoke() 执行图，断言最终状态是否符合预期。可以测试不同输入条件下，Agent 系统的执行路径和最终结果是否正确。

  def test_full_research_report_workflow():
       initial_input = {"messages": [HumanMessage(content="请研究一下 月球探索 的最新进展，并写一份摘要。")]}
       # 运行整个图
       final_state = app.invoke(initial_input)
  
       # 断言最终状态
       assert "messages" in final_state
       # 检查消息列表是否包含最终报告消息
       assert any("报告撰写完成" in msg.content for msg in final_state['messages'] if isinstance(msg, AIMessage))
       assert "final_report" in final_state
       assert len(final_state['final_report']) > 50 # 检查报告内容长度或特定关键词

单元测试和集成测试的结合能够有效地保证 Agent システム的质量和稳定性。可以使用 pytest 等测试框架来组织和运行测试。

8.4 模拟和桩 (Mocking and Stubbing)

在测试依赖外部服务（如 LLM API、数据库、网络搜索）的 Agent 时，为了避免真实调用带来的成本、延迟和不稳定性，可以使用模拟 (mocking) 和桩 (stubbing) 技术。

• 模拟 (Mocking): 替换真实的依赖对象，记录对其方法的调用，并控制其返回值。例如，模拟 LLM 的 invoke 方法，使其返回预设的回复。

• 桩 (Stubbing): 提供简化的、预设行为的替代品，用于在测试中满足依赖。例如，一个模拟的搜索工具函数，只返回固定的字符串结果，而不是执行真实的网络搜索。

Python 的 unittest.mock 模块或 pytest-mock 插件可以方便地实现模拟和桩。

# 使用 pytest-mock 模拟 search_the_web 工具
def test_researcher_agent_with_mocked_search(mocker):
    # 模拟 search_the_web 函数
    mock_search = mocker.patch('your_module.search_the_web', return_value="Mocked search results for the topic.")

    initial_state = ResearchReportState(messages=[], research_topic="虚拟主题", search_results="", final_report="")

    update = researcher_agent(initial_state)

    # 断言模拟的函数被调用，参数正确
    mock_search.assert_called_once_with("关于 虚拟主题 的最新信息")

    # 断言状态根据模拟结果更新
    assert update["search_results"] == "Mocked search results for the topic."

通过模拟外部依赖，您可以更专注于 Agent 内部逻辑和 Agent 之间的交互逻辑的测试。

第九章：部署 A2A Agent 的考量

将基于 Langchain 和 LangGraph 开发的 A2A Agent 系统从开发环境部署到生产环境需要考虑多个方面，包括可伸缩性、可靠性、安全性、监控和成本管理。

9.1 可伸缩性与性能

• 无状态 vs 有状态: LangGraph 是有状态的，其状态需要存储。选择合适的 State Checkpointer Backend 至关重要。内存 checkpointer 只适用于开发或单机场景。生产环境中需要使用支持持久化和并发访问的后端，如数据库（SQLite 文件或更强大的数据库如 PostgreSQL, MongoDB 等）或者专门的状态存储服务。

• 异步执行: 对于需要执行耗时操作（如网络请求、LLM 调用）的节点，考虑使用异步 I/O (asyncio) 来提高并发处理能力。 LangGraph 的 compile 方法支持 async 模式。 Agent 节点函数需要定义为 async def，并在内部使用 await 调用异步操作²。

• Agent 并行化: 在某些 A2A 架构中，可能需要并行执行多个子任务或调用多个 Agent。虽然 LangGraph 的核心执行模型是同步的图遍历，但可以通过将并行逻辑封装在一个节点中（例如，在该节点内部使用 asyncio.gather 并行调用多个 Runnable 或工具），或者设计 Agent 模式本身支持并行执行（如 MapReduce 模式）。

• LLM 调用吞吐量: LLM API 的调用可能是瓶颈。考虑使用支持高吞吐量的 LLM 模型和提供商。对于大量并发请求，需要合理设计 Agent 工作流，避免 LLM 的过度或重复调用。

9.2 可靠性与故障恢复

• 状态持久化 (Checkpointer): 这是确保可靠性的基础。如果 Agent 系统在执行过程中崩溃，通过持久化的状态，可以在恢复后从中断点继续执行，避免从头开始。选择可靠且具备备份机制的 checkpointer 后端。

• 错误处理: 在 Agent 节点函数中实现健壮的错误处理逻辑（try...except 块）。可以选择在节点内部处理错误并修改状态（例如，将任务状态标记为 failed），或者让错误冒泡，由 LangGraph 提供的错误处理机制来处理（例如，使用 add_edge(node, ERROR, error_handler_node) 将错误路由到专门的错误处理节点）。

• 幂等性: 设计 Agent 节点函数使其尽可能具有幂等性。这意味着多次执行同一个节点函数，如果输入状态没有改变，结果也是一样的。这有助于在重试或故障恢复时避免不一致。

9.3 安全性

• API 密钥管理: 绝不将 API 密钥硬编码在代码中。使用环境变量、Secret Manager 或其他安全的密钥管理服务来存储和访问密钥。

• 工具访问权限: Agent 调用的工具可能访问敏感数据或执行破坏性操作。确保 Agent 只能访问其完成任务所需的最小权限工具。在多 Agent 系统中，对 Agent 之间的通信和状态访问实施合适的访问控制。

• 输入验证和沙箱: 对用户输入和 Agent 之间传递的数据进行验证，防止注入攻击或恶意指令。考虑在沙箱环境中运行 Agent，限制其对外部系统的访问能力⁴。

• Agent 协议安全: 如果使用 A2A 协议进行跨系统通信，需要考虑身份认证（如 OAuth2, mTLS）、授权、传输加密（TLS/SSL）以及消息完整性验证¹²。

9.4 监控与日志

• 结构化日志: 使用结构化日志记录关键事件，如 Agent 开始/结束执行、状态变化、工具调用、错误等。这有助于在生产环境中进行集中式日志管理、搜索和分析。

• 应用监控: 使用应用性能监控(APM)工具来追踪 Agent 系统的性能指标，如请求延迟、错误率、吞吐量等。

• 业务指标: 监控与业务相关的指标，如任务完成率、用户满意度等，以评估 Agent 系统的实际效果。

• LangSmith (生产环境): LangSmith 不仅用于开发调试，也可以在生产环境中用于监控和跟踪。它可以帮助您理解生产流量下 Agent 系统的行为，发现潜在问题。

9.5 成本管理

• LLM 调用成本: LLM 调用是主要的成本来源。优化提示词、选择成本效益更高的模型、缓存重复的 LLM 调用、以及确保 Agent 不执行不必要的、昂贵的 LLM 调用是关键。

• 基础设施成本: 根据 Agent 系统的负载和并发需求，选择合适的計算資源和狀態存儲服務。异步处理和Agent并行化可以提高资源利用率。

• 工具使用成本: 某些工具（如付费 API）也可能产生费用。监控工具的使用情况，确保合理调用。

9.6 部署方式

基于 Python 和 LangGraph 的 A2A Agent 系统可以部署在多种环境中：

• Web 应用后端: 集成到 Flask、Django、FastAPI 等 Web 框架中，通过 API 接收请求并驱动 Agent 工作流。

• 容器化 (Docker): 将 Agent 系统打包到 Docker 容器中，便于在不同环境部署和管理。

• Serverless 函数: 对于请求量波动较大的场景，可以考虑将 Agent 工作流或部分节点部署为 Serverless 函数（如 AWS Lambda, Google Cloud Functions），按需付费。

• 云平台服务: 利用云平台提供的 Agent 或工作流编排服务（如 Google Cloud Agent Builder/Vertex AI Agent Framework，或其他平台提供的 Langchain/LangGraph 托管服务）。Google Cloud 的文档详细介绍了如何在 Vertex AI 上开发和部署 Langchain Agent²。

选择合适的部署方式取决于您的技术栈、运维能力、预算和伸缩性需求。

第十章：最佳实践和常见问题解答

构建健壮、可维护的 A2A Agent 系统需要遵循一些最佳实践。

10.1 最佳实践

• 模块化设计:

  • 将每个 Agent 的核心逻辑封装在独立的函数或类中。

  • 将工具定义清晰且功能单一。

  • 将复杂的 A2A 工作流分解为更小的子图，提高可理解性和复用性。

  • 状态定义应结构清晰，避免过于庞大和混乱。

• 明确的 Agent 职责: 每个 Agent 应该有清晰、单一的职责。避免 Agent 过于“全能”，导致其逻辑复杂、难以维护，并降低专业化优势。

• 规范的 Agent 间通信: 无论使用状态共享还是消息传递，建立一套规范的通信协议或数据格式。这有助于 Agent 之间准确理解信息，降低集成难度。例如，约定任务完成结果存放在状态的哪个字段，或者消息应该包含哪些关键信息。

• 充分利用状态管理: 将所有需要在 Agent 之间共享或需要长期记忆的信息都存储在 AgentState 中。合理设计状态结构和更新逻辑，确保状态的一致性和可追溯性。

• 可视化和追踪: 从项目一开始就集成 LangSmith 或其他追踪工具。可视化的工作流和详细的执行细节是调试和优化复杂 A2A 系统的杀手锏。

• 持续测试: 建立自动化测试流程，包括单元测试和集成测试。测试覆盖 Agent 节点的逻辑、条件边的跳转以及关键的工作流路径。

• 考虑异步和并行: 对于性能敏感的场景，评估 Agent 工作流中哪些部分可以并行执行，并利用 LangGraph 的异步能力和设计相应的 Agent 模式。

• 错误处理和容错: 仔细考虑各种可能的错误情况（LLM 调用失败、工具执行失败、状态不一致等），并在 Agent 节点和图级别实现相应的错误处理机制。

• 文档和注释: 详细记录每个 Agent 的功能、状态字段的含义、边和条件边的工作原理。清晰的代码注释和文档能够提高团队协作效率和系统的可维护性。

• 增量开发: 从简单的顺序协作模式开始，逐步引入更复杂的架构模式（如层级式、动态路由）， incrementally 构建和测试 A2A 系统。

10.2 常见问题解答 (FAQ)

• Q: 如何选择合适的 LLM 模型？ A: 选择 LLM 模型取决于您的需求、成本预算和延迟要求。

  • 对于需要强大推理、规划和遵循复杂指令的任务，考虑使用高性能模型（如 GPT-4 系列, Claude, Gemini Ultra）。

  • 对于成本敏感或只需处理简单任务的场景，可以考虑使用成本效益更高的模型（如 GPT-4o-mini, Claude 3 Haiku, Gemini Pro）。

  • 选择支持 function calling 或 tool use 能力的模型，可以简化 Agent 调用工具的实现。

  • 本地模型可以用于对数据隐私要求极高的场景。

• Q: 如何在 Agent 之间传递复杂数据结构？ A: 将复杂数据结构（如搜索结果列表、解析后的文档内容、中间计算结果等）直接存储在 AgentState 的相应字段中。确保这些数据类型在状态持久化时是可序列化的（通常 Python 基本类型、列表、字典、序列化后的对象都可以）。对于可变对象（列表、字典），在更新状态时注意合并逻辑¹。

• Q: 如何实现 Agent 之间的循环对话？ A: 使用 LangGraph 的条件边和状态管理。设计一个条件判断函数，检查对话是否需要继续（例如，根据用户输入、LLM 回复的结束标记或任务完成状态）。如果需要继续，条件边将流程导回生成回复的 Agent 节点或等待用户输入的节点（形成循环）。共享的消息列表 (MessagesState) 自动提供了完整的对话历史上下文。

• Q: 如何处理某个 Agent 节点执行失败的情况？ A:

  1. 节点内部处理: 在 Agent 节点函数中使用 try...except Block 捕获异常。在 except 中，更新状态以反映错误（例如，设置任务状态为 failed，在消息列表中添加错误消息），并返回更新后的状态。可以在错误处理后，根据状态决定是否需要重试、跳过任务或终止整个流程。

  2. 图级别处理: LangGraph 支持错误处理边。可以使用 add_edge(source_node, ERROR, error_handler_node) 将从 source_node 抛出的错误路由到一个专门的错误处理节点，在该节点中进行统一的日志记录、通知或尝试恢复。

• Q: 如何在 LangGraph 中实现并行执行？ A: LangGraph 的核心执行是顺序沿着边进行的。要实现并行，可以在一个 Agent 节点内部使用 Python 的异步编程能力（如 asyncio.gather）并行调用多个 LLM、工具或其他 Runnable。或者，设计 Agent 工作流时，让一个 Agent 将多个子任务分配出去，由不同的 LangGraph 实例或独立的 Worker 进程/线程并行处理，最后再由另一个 Agent 节点收集和汇总结果。

• Q: 如何将人类反馈集成到 A2A 工作流中 (Human-in-the-Loop)? A: 在 LangGraph 图中，可以将一个节点设计为“人类审核节点”。当 Agent 工作流需要人类干预时，流程转移到此节点。此节点可以暂停执行，将当前状态（包括 Agent 的最新思考、待决策事项等）发送给人类用户界面。接收到人类反馈后，将反馈添加到状态中，并根据反馈结果通过条件边决定流程的下一步（例如，返回某个 Agent 进行修改，或者直接进入下一个阶段）。 LangGraph 的状态持久化特性对于实现长期的人类参与至关重要。

• Q: LangChain Agent 和 LangGraph 节点有什么关系？ A: LangGraph 节点可以是一个简单的 Python 函数，也可以更高级地封装一个完整的 LangChain AgentExecutor。将 LangChain AgentExecutor 作为 LangGraph 节点可以充分利用 LangChain Agent 的 ReAct 推理、工具调用、记忆管理等功能。在 LangGraph 中定义 Agent 节点时，您可以选择直接在函数中编写 Agent 逻辑（调用 LLM、工具），或者创建 Langchain AgentExecutor 实例作为节点。

• Q: 如何在 LangGraph 中管理多个独立的会话？ A: 使用 LangGraph 的状态持久化 (checkpointer)。每次 invoke 或 stream 调用时，在 config 中传入一个唯一的 thread_id。 LangGraph 会根据 thread_id 加载或保存对应的会话状态。这使得同一个 LangGraph 应用实例可以同时处理多个独立的 Agent 会话。

总结与展望

本教程作为一份详尽的 Langchain + LangGraph A2A 开发指南，从 Agent 和多 Agent 系统的基础概念出发，深入剖析了 LangGraph 的核心 구성、状态管理和控制流机制，并通过研究报告生成和任务分解协作两个具体的 A2A 示例，展示了如何使用 LangGraph 构建实际的多 Agent 工作流。我们还探讨了不同的 A2A 架构模式、Agent 间的通信机制、调试与测试方法以及生产部署的关键考量。

通过学习本教程，您应该已经掌握了：

• Langchain 和 LangGraph 在 A2A 开发中的作用和优势。

• LangGraph 的图结构、节点、边和状态管理的核心概念。

• 使用 Python 和 LangGraph 构建基础 Agent 和复杂 A2A 工作流的方法。

• 实现 Agent 之间信息传递和协作的技术。

• 使用 LangSmith 进行可视化调试和追踪。

• A2A 系统开发、测试和部署的最佳实践和注意事项。

A2A 系统是构建更智能、更自主、更具协作能力的 AI 应用的关键发展方向。 Langchain 和 LangGraph 作为领先的框架，为开发者提供了强大的工具集。 随着 LLM 能力的不断提升和 Agent 技术的发展，未来的 A2A 系统将能够处理更加复杂、开放域的任务，实现更紧密的 Agent 协同，并在更多领域（如企业自动化、科学研究、智能制造、创意产业）展现出巨大的潜力。

希望本教程能为您在 A2A Agent 开发的旅程中提供坚实的基础和实用的指导。 现在，是时候动手尝试，构建您自己的 Agent 协作系统了！

Agent-to-Agent (A2A) 协议的标准化将进一步促进跨平台、跨厂商的 Agent 协作生态系统的发展，支持去中心化 Agent 网络的构建，未来 Agent 可能会像 Web 服务一样通过开放协议进行交互¹²⁵。

不断探索、实践和创新，用 Agent 技术构建更智能的未来！