当传统的AI代理将所有能力捆在一个黑盒里时,LangGraph正通过“思考与行动分离”的架构,重新定义可维护的AI工作流。

在构建基于大语言模型的应用时,我们常陷入一个直观的设计:一个全能的“代理”,它既能理解问题,又能调用工具解决问题。这就像创造一个超人,但问题也随之而来——当这个“超人”需要增加新能力、排查故障或优化性能时,修改往往牵一发而动全身。

LangGraph提出了不同的思路,它将“思考” 与“行动” 解耦,通过有向图清晰的流程控制,带来前所未有的模块化与可维护性。今天,我们深入探讨这一设计的精髓。

一、传统架构的困境:捆绑的代价

典型的AI代理设计将LLM推理与工具调用紧密耦合:

# 传统代理的典型结构
class AllInOneAgent:
    def __init__(self, llm, tools):
        self.llm = llm
        self.tools = tools
    
    def run(self, query):
        # LLM决定行动
        decision = self.llm.decide(query, self.tools)
        
        if decision.action == "use_tool":
            # 代理自己执行工具
            result = self.execute_tool(decision.tool_name, decision.args)
            return self.llm.process_result(result)
        else:
            return self.llm.generate_response()

这种设计的挑战很明显:逻辑交织导致测试困难、关注点混杂使得扩展复杂、状态管理变得隐晦不清。当系统需要新增一个工具或修改现有工具行为时,你不得不深入代理的核心逻辑。

二、LangGraph的解构:清晰的职责分离

LangGraph通过有向图将工作流解构为独立的节点,下图清晰地展示了这一架构中“思考”与“行动”的分离与协作:

关键架构转变

  • LLM节点:纯思考与决策,根据对话历史和当前状态,决定下一步行动

  • 工具节点:纯执行与反馈,接收标准化指令,执行并返回结构化结果

三、架构实现:代码中的分离哲学

1. 定义状态与工具

from typing import TypedDict, List, Annotated, Literal
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langchain.tools import tool
from langchain_openai import ChatOpenAI
from langchain_core.messages import AIMessage, ToolMessage

# 定义清晰的状态结构
class AgentState(TypedDict):
    messages: Annotated[List, add_messages]  # 完整的对话上下文
    needs_tool: bool  # 明确的决策标志位
    tool_result: dict  # 工具执行结果的专用存储

# 定义专用工具(而非通用函数)
@tool
def search_database(query: str) -> dict:
    """执行数据库查询,返回结构化数据。"""
    # 实际的数据库调用逻辑
    return {"results": [...], "source": "database"}

@tool  
def call_external_api(params: dict) -> dict:
    """调用外部API服务。"""
    # API调用逻辑
    return {"status": "success", "data": {...}}

2. 实现思考节点:纯决策逻辑

# 初始化LLM并绑定工具
llm = ChatOpenAI(model="gpt-4", temperature=0)
tools = [search_database, call_external_api]
llm_with_tools = llm.bind_tools(tools)

def thinking_node(state: AgentState) -> dict:
    """思考节点:分析上下文并决策,不执行具体操作"""
    messages = state["messages"]
    
    # LLM分析当前状态
    ai_message = llm_with_tools.invoke(messages)
    
    # 明确决策:是否需要工具?
    needs_tool = hasattr(ai_message, 'tool_calls') and ai_message.tool_calls
    
    # 更新状态
    updates = {"messages": [ai_message], "needs_tool": needs_tool}
    
    # 如果有工具调用,提取参数到专用字段
    if needs_tool:
        updates["tool_call"] = ai_message.tool_calls[0]
    
    return updates

3. 实现工具节点:纯执行逻辑

def action_node(state: AgentState) -> dict:
    """行动节点:专一执行工具调用"""
    tool_call = state["tool_call"]
    tool_name = tool_call["name"]
    tool_args = tool_call["args"]
    
    # 根据工具名分发给具体实现
    if tool_name == "search_database":
        result = search_database.invoke(tool_args)
    elif tool_name == "call_external_api":
        result = call_external_api.invoke(tool_args)
    else:
        result = {"error": f"未知工具: {tool_name}"}
    
    # 创建标准化工具消息
    tool_message = ToolMessage(
        content=str(result),
        name=tool_name,
        tool_call_id=tool_call.get("id", "1")
    )
    
    return {
        "messages": [tool_message],
        "tool_result": result,
        "needs_tool": False  # 重置标志位
    }

4. 构建完整工作流图

# 创建图构建器
workflow = StateGraph(AgentState)

# 添加独立节点
workflow.add_node("思考", thinking_node)
workflow.add_node("行动", action_node)
workflow.add_node("回答", lambda state: state)  # 最终回答节点

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

# 定义清晰的条件路由
def route_decision(state: AgentState) -> Literal["行动", "回答", END]:
    """路由决策:根据思考结果决定下一步"""
    
    # 检查是否有工具调用请求
    if state.get("needs_tool"):
        return "行动"
    
    # 检查是否已生成最终答案
    last_msg = state["messages"][-1] if state["messages"] else None
    if isinstance(last_msg, AIMessage) and not last_msg.tool_calls:
        return "回答"
    
    # 默认继续思考
    return "思考"

# 连接节点
workflow.add_conditional_edges(
    "思考",
    route_decision,
    {"行动": "行动", "回答": "回答", END: END}
)

workflow.add_edge("行动", "思考")  # 行动后返回思考
workflow.add_edge("回答", END)     # 生成回答后结束

# 编译工作流
app = workflow.compile()

四、架构优势:不只是代码组织

1. 可观测性

每个节点的输入输出明确记录在状态中,调试时可以直接检查每个决策点的状态:

# 调试时追踪状态变化
for step in app.stream(initial_state):
    node_name, node_state = list(step.items())[0]
    print(f"节点 [{node_name}] 状态: {node_state.keys()}")
    # 可以看到:思考节点输出 needs_tool=True,行动节点输出 tool_result=...

2. 独立测试

每个节点可以独立测试,无需启动完整LLM:

# 独立测试行动节点
def test_action_node():
    test_state = {
        "tool_call": {"name": "search_database", "args": {"query": "测试"}},
        "messages": [],
        "needs_tool": True
    }
    result = action_node(test_state)
    assert "tool_result" in result
    print("行动节点测试通过")

3. 动态扩展

新增工具只需扩展行动节点,不影响思考逻辑:

# 扩展行动节点支持新工具
def extended_action_node(state: AgentState) -> dict:
    tool_name = state["tool_call"]["name"]
    
    # 原有工具...
    if tool_name in ["search_database", "call_external_api"]:
        # ... 原有逻辑
    
    # 新增工具
    elif tool_name == "new_tool":
        return execute_new_tool(state)
    
    return {"error": "未知工具", "messages": []}

4. 流程定制

通过调整图结构,轻松创建不同的工作流模式:

4. 流程定制
通过调整图结构,轻松创建不同的工作流模式:

五、总结:从“全能代理”到“专业团队”

LangGraph通过“思考与行动分离”的架构,将AI应用开发从创造单个全能代理,转向组建专业协作团队

维度 传统代理 LangGraph架构
职责分配 集中混合 清晰分离
可维护性 修改影响大 独立修改节点
可测试性 集成测试困难 节点独立测试
可观测性 黑盒决策 透明状态流转
扩展性 需修改核心 增删节点即可

这种架构的真正价值在于工程化当AI应用从原型走向生产环境时,清晰的分层、明确的职责和可维护的代码结构变得至关重要。LangGraph不仅是一个技术框架,更是一种架构哲学——承认复杂性,然后通过良好的设计管理它。

Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐