2. 使用LangGraph的状态图实现Agent间的消息传递

任务说明

本案例将使用 LangGraph 构建一个包含两个 Agent 的状态图，实现消息在 Agent 间的顺序传递。两个 Agent 分别扮演“提问者”和“回答者”，通过共享状态中的消息列表进行交互，完成一轮简单的顺序对话。

环境依赖

需要安装 langgraph 和 langchain-core。本案例使用模拟的 Agent 响应函数，不依赖外部大模型 API，便于独立运行验证。

pip install langgraph langchain-core

核心实现

状态图的核心是定义一个共享状态结构，包含消息列表字段。两个 Agent 节点分别向消息列表追加消息，并通过条件边控制对话轮次，实现顺序传递。

from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langchain_core.messages import AIMessage, HumanMessage

# 1. 定义共享状态结构
class AgentState(TypedDict):
    messages: Annotated[List, add_messages]  # 使用 add_messages 实现消息追加合并

# 2. 定义两个 Agent 节点（模拟响应逻辑）
def questioner_agent(state: AgentState) -> dict:
    """提问者 Agent：根据当前对话历史生成下一个问题"""
    # 模拟：如果还没有提问，则提出第一个问题
    if len(state["messages"]) == 0:
        return {"messages": [AIMessage(content="什么是LangGraph的状态图？")]}
    # 如果已经收到回答，则提出追问
    return {"messages": [AIMessage(content="状态图如何管理消息传递？")]}

def answerer_agent(state: AgentState) -> dict:
    """回答者 Agent：对最后一个问题给出回答"""
    last_message = state["messages"][-1]
    # 模拟回答逻辑
    return {"messages": [AIMessage(content=f"关于「{last_message.content}」，答案是：状态图通过共享状态和边实现消息流转。")]}

# 3. 构建状态图
def build_agent_graph():
    # 创建状态图，绑定状态结构
    graph = StateGraph(AgentState)
    
    # 添加节点
    graph.add_node("questioner", questioner_agent)
    graph.add_node("answerer", answerer_agent)
    
    # 设置入口节点
    graph.set_entry_point("questioner")
    
    # 添加边：提问者 -> 回答者
    graph.add_edge("questioner", "answerer")
    
    # 条件边：回答者根据消息数量决定是否继续提问
    def should_continue(state: AgentState) -> str:
        # 如果消息数量小于4（即完成一轮问答），则回到提问者继续
        if len(state["messages"]) < 4:
            return "questioner"
        return "end"
    
    graph.add_conditional_edges("answerer", should_continue, {
        "questioner": "questioner",
        "end": END
    })
    
    return graph.compile()

# 4. 运行入口
if __name__ == "__main__":
    # 初始化空消息列表
    initial_state = {"messages": []}
    
    # 编译并运行图
    app = build_agent_graph()
    result = app.invoke(initial_state)
    
    # 打印完整对话
    print("=== 对话记录 ===")
    for msg in result["messages"]:
        print(f"{msg.type}: {msg.content}")

运行与验证

直接运行上述脚本即可看到两个 Agent 的完整对话过程。程序会依次输出提问者的问题和回答者的回复，共两轮交互（4条消息）。

验证通过条件：控制台输出包含 4 条消息，且消息顺序为：提问 → 回答 → 提问 → 回答。第一条提问内容为“什么是LangGraph的状态图？”，第二条提问为“状态图如何管理消息传递？”。

关键机制说明

本案例的核心机制在于 add_messages 归约器：它使得每个节点返回的新消息会自动追加到状态中的 messages 列表，而不是覆盖旧消息。条件边 should_continue 通过检查消息总数控制对话轮次，当消息数达到 4 时终止流程，从而形成完整的顺序对话闭环。