Langgraph 项目代码编写步骤（Router 结构）

zadyd

258人浏览 · 2026-03-20 17:05:40

zadyd · 2026-03-20 17:05:40 发布

本人最近在编写一个 router 结构的 agent，以下讨论仅仅适用于 router 结构

1. State (状态定义) $→\rightarrow$ 2. Bottom Nodes (底层功能节点) $→\rightarrow$ 3. Router Node (路由节点/逻辑) $→\rightarrow$ 4. Graph Construction (图构建)

为什么是这个顺序？

State 是契约：所有节点（包括路由和功能节点）都需要读写 State。不先定义 State，就不知道节点函数的入参和返回值是什么。
Nodes 是目的地：路由器的作用是将流量分发给具体的功能节点。如果你还没写好功能节点，路由器就没有“目的地”可指。
Router 是决策者：它依赖于 State 中的历史信息，并决定流向哪些 Nodes。

完整开发步骤指南

下面我将通过一个具体的例子（区分“通用聊天”和“搜索工具”）来演示完整的开发流程。

第一步：定义 State (状态)

这是整个图的“内存”结构。对于路由架构，通常需要由系统默认的 messages 和可能的 next_node（可选，取决于路由实现方式）组成。

from typing import Annotated, Literal, TypedDict
from langgraph.graph.message import add_messages

# 1. 定义状态
class AgentState(TypedDict):
    # messages: 用于存储对话历史，add_messages 表示追加而不是覆盖
    messages: Annotated[list, add_messages]
    # (可选) 如果你的路由节点不仅返回路径，还写入决策结果，可以在这里加字段
    # intent: str

第二步：编写 Bottom Nodes (底层功能节点)

这些是实际干活的节点（Tools 或子 Agent）。路由将把请求转发给它们。

from langchain_core.messages import AIMessage

# 2.1 节点 A：处理通用闲聊
def general_chat_node(state: AgentState):
    # 模拟 LLM 调用
    return {"messages": [AIMessage(content="我是通用聊天助手，很高兴见到你。")]}

# 2.2 节点 B：处理搜索/具体任务
def search_tool_node(state: AgentState):
    # 模拟工具调用
    query = state["messages"][-1].content
    result = f"关于 '{query}' 的搜索结果..."
    return {"messages": [AIMessage(content=result)]}

第三步：编写 Router (路由逻辑)

这是核心部分。路由通常有两种实现方式：

作为独立节点：先由一个 LLM 节点分析意图，更新 State，然后由条件边读取 State。
作为条件边逻辑：直接在边（Edge）的函数里调用 LLM 进行判断（推荐更轻量级的做法）。

这里演示结构化输出 (Structured Output) 的方式，这是目前最稳健的路由写法。

from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field

# 定义路由的数据模型
class RouteDecision(BaseModel):
    """根据用户输入判断下一步去向"""
    next: Literal["general_chat", "search_tool"] = Field(
        ..., description="如果是闲聊选 general_chat，如果是查询资料选 search_tool"
    )

# 初始化 LLM
llm = ChatOpenAI(model="gpt-4o")
# 绑定结构化输出，强制 LLM 返回 JSON 格式的决策
router_llm = llm.with_structured_output(RouteDecision)

# 3. 定义路由逻辑函数（这不是一个节点，而是用于 conditional_edges 的函数）
def route_decision(state: AgentState):
    last_message = state["messages"][-1]
    decision = router_llm.invoke([last_message])
    
    # 返回目标节点的名称
    if decision.next == "general_chat":
        return "general_chat_node"
    else:
        return "search_tool_node"

第四步：构建 Graph (组装)

将 State、Nodes 和 Router 逻辑组装在一起。

from langgraph.graph import StateGraph, END

# 4.1 初始化图
workflow = StateGraph(AgentState)

# 4.2 添加节点
workflow.add_node("general_chat_node", general_chat_node)
workflow.add_node("search_tool_node", search_tool_node)

# 4.3 设置入口点和路由
# 注意：这里我们不需要一个专门的 "Router Node"，而是直接从入口(START)开始路由
# 也可以先加一个 "Orchestrator" 节点，然后从那个节点路由
workflow.set_conditional_entry_point(
    route_decision, # 路由逻辑函数
    path_map={      # 映射关系：函数返回值 -> 节点名称
        "general_chat_node": "general_chat_node",
        "search_tool_node": "search_tool_node"
    }
)

# 4.4 设置出口
# 功能节点执行完后，通常指向 END，或者回到某个主节点
workflow.add_edge("general_chat_node", END)
workflow.add_edge("search_tool_node", END)

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

第五步：测试运行

# 测试 1：闲聊
inputs = {"messages": [("user", "你好")]}
for output in app.stream(inputs):
    for key, value in output.items():
        print(f"Node '{key}': {value}")

# 测试 2：搜索
inputs = {"messages": [("user", "今天的天气怎么样")]}
for output in app.stream(inputs):
    for key, value in output.items():
        print(f"Node '{key}': {value}")