砸开Agent黑盒！LangGraph从零搭白盒ReAct Agent

2025-11-18 15:49:49
文章摘要
用AgentExecutor搭AI Agent时，你是不是也遇到过这些糟心问题？ → 工具调用了几次全靠猜，调试像“蒙眼开车”； → 危险操作前想加人工确认，却改不动封装逻辑； → 想加“重试”“总结”功能，对着黑盒无从下手。 别慌！本文手把手教你用LangGraph“砸开”黑盒，从零构建透明可控的白盒ReAct Agent，还会集成LangSmith实现全流程调试——配套代码已开源。_

   一、为啥非要用LangGraph？—— 黑盒Agent的致命伤

之前我们用LangChain的AgentExecutor快速搭出了能调工具、记历史的Agent，一行代码搞定所有：

   

agent_executor = AgentExecutor(agent=agent, tools=tools)

   但这套“懒人方案”的代价是——全流程不可控。它封装了完整的ReAct循环（思考→行动→观察），却把关键逻辑藏得严严实实：

● ❌ 中间调用工具的次数、输入输出全看不到；

● ❌ “删除文件”“调用支付接口”等危险操作，没法插人工确认；

● ❌ 想加超时中断、失败重试、结果总结，根本找不到修改入口。


而LangGraph的核心价值，就是把这个黑盒“拆成透明零件”——它不替代LangChain，而是用代码实现“可视化工作流”，让你亲手定义Agent的每一步执行逻辑，灵活度直接拉满。


二、LangGraph入门：3要素+4步法拼出流程

用过Dify、Coze的同学都知道，搭Agent靠“拖拽节点+连线条”。LangGraph就是这套逻辑的代码版，核心只需要搞懂3个东西，再按4步拼装就行。

1. 核心三要素：State、Node、Edge

这三者就像“白板+工人+工头”，配合起来完成整个工作流：

State（状态）

类比“共享白板”，存储对话历史、工具结果等所有数据，所有节点都能读写。

Node（节点）

类比“工人”，是执行具体任务的函数（如调用LLM、执行工具），干完活把结果写回State。

Edge（边）

类比“工头”，定义节点执行顺序（固定跳转或条件分支，比如“成功去A，失败去B”）。


2. 四步拼装法：从0写一个极简流程

用“计数累加”的极简案例，带你吃透完整流程（代码可直接复制运行）：



from typing import TypedDict
from langgraph.graph import StateGraph, START, END

# 1. 定义State：决定白板上有什么字段（这里只存一个count）
class State(TypedDict):
    count: int

# 2. 编写Node：两个工人，负责计数+1
def node_a(state: State):
    print(f"[节点A] 收到计数：{state['count']}")
    return {"count": state["count"] + 1}  # 结果写回白板

def node_b(state: State):
    print(f"[节点B] 收到计数：{state['count']}")
    return {"count": state["count"] + 1}

# 3. 添加Node到图中：创建画布并放好工人
workflow = StateGraph(State)
workflow.add_node("A", node_a)  # 节点命名+对应函数
workflow.add_node("B", node_b)

# 4. 用Edge连线：定义执行顺序（START→A→B→END）
workflow.add_edge(START, "A")
workflow.add_edge("A", "B")
workflow.add_edge("B", END)

# 编译成可运行应用
app = workflow.compile()

# 执行：传入初始状态{count:1}
print("---开始执行---")
result = app.invoke({"count": 1})
print("最终结果:", result)  # 输出 {'count': 3}


运行后会打印节点执行过程，最终计数从1累加到3——这就是LangGraph的核心逻辑：状态驱动节点，节点通过边串联。

3. 可视化与调试：LangSmith必须安排上

上面的代码能跑，但执行过程还是“文字日志”。想直观看到流程、定位问题，必须用LangSmith——它是LangGraph的“飞行记录仪”，能记录每一步的输入输出、耗时、Token消耗。

LangSmith不是可选工具，而是LangChain/LangGraph开发的“标配”，就像Chrome DevTools之于前端工程师。


快速接入LangSmith步骤：

1.  注册账号：去LangSmith官网，用GitHub/Google/邮箱注册；

2.  创建API Key：在控制台生成API Key，保存好；

3.  配置环境变量：在代码开头加3行配置，其余代码不变：

   

import os
# 启用追踪功能
os.environ["LANGCHAIN_TRACING_V2"] = "true"
# 项目名（自定义，方便分类）
os.environ["LANGCHAIN_PROJECT"] = "LangGraph-Demo"
# 替换成你的LangSmith API Key
os.environ["LANGCHAIN_API_KEY"] = "你的API_KEY"


   运行代码后，打开LangSmith控制台，就能看到完整的执行轨迹：

● ✅ 自动绘制执行流程图（START→A→B→END）；

● ✅ 点击节点看详细输入输出、耗时；

● ✅ 标红异常节点，直接看错误堆栈；

● ✅ 统计Token消耗和成本估算。


三、核心实战：用LangGraph替换AgentExecutor

现在进入重头戏——用LangGraph实现标准ReAct循环，彻底替代AgentExecutor的黑盒逻辑。目标：构建一个能“判断是否需要调用工具”的AI Agent。

1. 准备工作：LLM+工具配置

先配置大模型和工具（这里用“获取天气”模拟工具，实际可替换成搜索、数据库等）：



from langchain.schema import HumanMessage
from langchain.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, MessagesState, END, START
from langgraph.prebuilt import ToolNode

# 1. 配置LLM（这里用DeepSeek，可替换成GPT/文心一言）
llm = ChatOpenAI(
    model="deepseek-chat",
    api_key="你的DeepSeek API_KEY",
    base_url="https://api.deepseek.com"
)

# 2. 定义工具（模拟获取天气）
@tool
def get_weather(location: str):
    """获取指定城市的天气信息，参数为城市名"""
    return f"{location}当前天气：23℃，晴，风力2级"

tools = [get_weather]
llm_with_tools = llm.bind_tools(tools)  # 让LLM学会调用工具


2. 拆解ReAct循环：3个节点+1个条件判断

ReAct的核心是“思考→行动→观察→再思考”，我们用3个节点实现，再加1个条件判断控制循环：



# 节点1：Thought（思考）—— 调用LLM判断是否需要工具
def call_model(state: MessagesState):
    # 传入当前对话历史，让LLM决策
    response = llm_with_tools.invoke(state['messages'])
    return {"messages": [response]}  # 结果写回State

# 节点2：Action+Observation（行动+观察）—— 执行工具
tool_node = ToolNode(tools)  # LangGraph内置工具节点，自动执行tool_calls

# 节点3：Loop Controller（循环控制）—— 判断是否继续
def should_continue(state: MessagesState):
    last_msg = state["messages"][-1]
    # 有工具调用需求→去执行工具；无→直接返回结果
    if hasattr(last_msg, "tool_calls") and len(last_msg.tool_calls) > 0:
        return "tools"
    return END


3. 搭建循环图：用条件边实现动态路由

关键用add_conditional_edges实现“动态跳转”，这是LangGraph的核心亮点：



# 1. 创建基于MessagesState的工作流（自带对话历史存储）
workflow = StateGraph(MessagesState)

# 2. 添加节点
workflow.add_node("agent", call_model)  # 思考节点
workflow.add_node("tools", tool_node)   # 工具节点

# 3. 连线：START→思考节点
workflow.add_edge(START, "agent")

# 4. 条件边：思考节点→决定下一步（去工具节点或结束）
workflow.add_conditional_edges(
    "agent",  # 起点节点
    should_continue,  # 决策函数
    {
        "tools": "tools",  # 决策返回"tools"→去工具节点
        END: END # 决策返回END→流程结束
    }
)

# 5. 工具节点执行完→回到思考节点，形成循环
workflow.add_edge("tools", "agent")

# 编译应用（记得加LangSmith配置）
app = workflow.compile()


4. 运行测试：黑盒逻辑彻底透明



if __name__ == '__main__':
    # 测试1：需要调用工具（问天气）
    print("---测试工具调用---")
    result = app.invoke({"messages": [HumanMessage(content="北京天气如何？")]})
    print("AI回答:", result['messages'][-1].content)

    # 测试2：不需要调用工具（打招呼）
    print("\n---测试直接回答---")
    result = app.invoke({"messages": [HumanMessage(content="你好！")]})
    print("AI回答:", result['messages'][-1].content)


运行后去LangSmith看轨迹，会清晰看到：

● 问天气时：agent→tools→agent，完整执行ReAct循环；

● 打招呼时：agent直接返回，流程无冗余步骤。

此时的Agent不再是“魔法”，每一步都透明可控——想在工具执行前加人工确认？只需在tool_node前加一个“审批节点”，逻辑轻松插入。


四、进阶：给Agent加记忆+安全注入系统提示

基础版Agent还缺两个生产级能力：多轮记忆和安全的系统提示。LangGraph实现起来超简单，只需几行代码。

1. 安全注入系统提示：不污染历史记录

直接把SystemMessage写入State会导致多轮对话重复注入，污染历史。正确做法是：调用LLM时临时拼接，不写入状态：



# 定义系统提示
sys_prompt = "你是专业助手，调用工具前先确认参数是否完整，回答简洁明了。"

# 改造思考节点：临时拼接系统提示
def call_model(state: MessagesState):
    # 仅本次LLM调用用，不写入State
    messages_for_llm = [SystemMessage(content=sys_prompt)] + state["messages"]
    response = llm_with_tools.invoke(messages_for_llm)
    return {"messages": [response]}


2. 启用持久化记忆：仅需两行代码

LangGraph的记忆靠checkpointer实现，不用手动管理对话历史，编译时加一句就行：



from langgraph.checkpoint.memory import MemorySaver

# 1. 初始化记忆存储
memory = MemorySaver()

# 2. 编译时启用记忆
app = workflow.compile(checkpointer=memory)


3. 多轮交互测试：会话隔离+记忆持久化



if __name__ == '__main__':
    # 用thread_id区分不同用户（会话隔离）
    config = {"configurable": {"thread_id": "user_001"}}
    
    while True:
        user_input = input("\n你：")
        if user_input.lower() == "quit":
            break
        # 每次仅传入新问题，记忆自动关联
        result = app.invoke(
            {"messages": [HumanMessage(content=user_input)]},
            config=config
        )
        print(f"AI：{result['messages'][-1].content}")


测试时问“北京天气”，再问“明天呢？”，Agent会自动关联上下文——记忆功能完美生效，且不同thread_id的会话互不干扰。


总结：LangGraph核心知识点回顾

1. 三要素：State（状态存储）、Node（任务执行）、Edge（流程控制）；

2. 核心能力：ReAct循环白盒化、条件路由、持久化记忆、系统提示安全注入；

3. 调试神器：LangSmith全流程追踪，定位问题效率翻倍；

4. 优势：兼容LangChain生态，灵活度远超AgentExecutor。



声明：该内容由作者自行发布，观点内容仅供参考，不代表平台立场；如有侵权，请联系平台删除。