LangChain+LangGraph实战：构建工具调用、记忆与人工审批齐备的Agent

写Agent不能只停留在“模型调一下函数、返回一段 JSON、再拼个回复”的阶段。真正能上生产的 Agent，通常要同时解决工具调用、会话记忆、人工确认、权限控制这些问题。本文从 LangChain + LangGraph 的新体系出发，不是讲概念，而是直接给一套能在本机跑起来的完整结构：工具注册、短长期记忆、人机协同、安全护栏，全部串在同一个 Agent 里。也会说明每个环节适合什么业务场景、验证方式、以及常见坑点。

1. LangChain 完整体系核心能力速览

把这个体系当成一个工程框架来看待，最值得关注的不是单个 API，而是它把 Agent 的各个能力模块化到了 Graph 节点里：模型调用、工具执行、记忆读写、人工审批、输出校验，都能单独控制。

从材料给出的信息看，这个体系的重点是解决“简单 Agent 不可控”的问题：单轮调用在 demo 里很好看，一进入真实业务就会出现状态丢失、越权操作、敏感信息泄露、无法人工干预等事故。LangChain 新版把这些能力收敛成标准组件，配合 LangGraph 的图执行机制来落地。

2. 为什么“简单 Agent”不够用

先看一个最常用的简版 Agent 实现：

python复制from openai import OpenAI

client = OpenAI()

def call_llm(prompt):
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

这种代码能跑，也只是能跑。你很快会遇到几个实际问题：

• 没有工具调用协议：模型输出什么格式、参数怎么填、工具报错了怎么继续，全部要自己写 if-else。
• 没有短期记忆：用户上一轮说“查一下北京天气”，这一轮问“那上海呢”，模型不知道“那”指代什么。
• 没有长期记忆：用户偏好、历史订单、历史工单全都拿不到，哪怕数据库里已经有 10 万条历史数据。
• 没有人机协同：模型认为需要“删除订单”就直接执行了，中间没有任何人来确认。
• 没有安全护栏：模型被提示词注入后可能去调用不该调用的工具，或者把系统 prompt、内部配置直接输出给用户。

LangChain 新版本的价值就在这里：把上述能力从“自己手搓”变成可组装模块。尤其是 LangGraph 的图状态机制，让 Agent 从“一次性循环”变成“可在任意节点暂停、恢复、分支执行”的流程系统。工具调用、记忆、人工审批、护栏，都只是图里的节点。

3. 环境准备与项目初始化

3.1 安装依赖

建议使用虚拟环境，避免依赖冲突。

bash复制python -m venv .venv
source .venv/bin/activate      # Windows 用 .venv\Scripts\activate

安装核心依赖：

bash复制pip install -U langgraph langchain langchain-openai langchain-core langchain-chroma

如果后续要包装 HTTP 接口，额外安装：

bash复制pip install fastapi uvicorn

3.2 配置模型

这个体系支持云端模型和本地模型。云端模型需要准备 API Key：

bash复制export OPENAI_API_KEY="sk-xxxx"

本地模型可以借助 Ollama 运行，并把 LangChain 的模型实例换成 Ollama 接口。对显存的要求取决于模型参数规模，比如在 8GB 显存附近跑量化后的小参数模型比较稳妥，实际占用要以本机测试为准。

3.3 验证安装

python复制from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
print(llm.invoke("1+1=").content)

能正常输出结果即表示环境可用。此步骤可以与后续功能测试合并执行，不必单独跑通一套。

4. 工具调用：让 Agent 从“聊天”变成“干活”

4.1 注册一个工具

LangChain 使用 @tool 装饰器定义工具。关键在于写清楚函数签名和 docstring，模型会依靠这些信息决定是否调用。

python复制from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """查询指定城市的实时天气。输入参数为城市名称，例如“上海”。"""
    # 实际项目替换为真实天气服务接口
    return f"{city} 今日 24 度，多云，东南风 3 级"

4.2 创建可调用工具的 Agent

使用 LangGraph 的 create_react_agent 预设结构：

python复制from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
tools = [get_weather]

agent = create_react_agent(model=llm, tools=tools)

result = agent.invoke({"messages": [("human", "上海今天适合跑步吗？")]})
print(result["messages"][-1].content)

运行后会看到模型先输出工具调用，再读取工具返回结果，最后整合成自然语言回答。通过这一段代码就能确认：模型是否自动选择了工具、工具参数是否解析正确、工具返回内容是否被模型正确使用。

4.3 多工具选择测试

真实场景中通常要给 Agent 提供多个工具。把工具列表扩展一下：

python复制@tool
def get_humidity(city: str) -> str:
    """查询指定城市的相对湿度。输入参数为城市名称。"""
    return f"{city} 当前湿度 62%"

tools = [get_weather, get_humidity]

再让模型回答“上海湿度多少、温度多少”。判断标准是：模型会连续调用两个工具，并把结果合并成最终回答。如果模型只调了一个工具，通常是工具描述不清晰或模型版本对 tool calling 支持不稳定。

5. 长短记忆：Agent 状态从“单轮”到“持久”

5.1 短期记忆：会话内上下文

短期记忆的本质是保留会话历史。LangGraph 中用 Checkpoint 保存消息状态，传入 thread_id 即可区分不同会话。

python复制from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()

agent = create_react_agent(
    model=llm,
    tools=tools,
    checkpointer=memory,
)

config = {"configurable": {"thread_id": "user-001"}}

agent.invoke(
    {"messages": [("human", "查一下北京的天气")]},
    config=config,
)

agent.invoke(
    {"messages": [("human", "那上海呢？")]},
    config=config,
)

第二次调用时，Agent 能从已有消息历史中理解“那”指的是天气查询。这就是短期记忆的典型作用。注意 MemorySaver 只存在内存里，进程重启后会丢失。生产环境建议使用 PostgreSQL 或 Redis 作为持久化 Checkpoint。

5.2 长期记忆：用户画像与业务偏好

长期记忆是跨会话、跨进程的。它的核心不是把对话记录全部堆给模型，而是提取关键信息并存储到外部系统。

做法通常有两种：

• 结构化画像：把用户偏好、订单状态、审批权限写入数据库或 JSON 文件。
• 语义记忆：把历史对话切片向量化，存入向量数据库，下次按语义相似度召回。

这里给出一个简单的长期记忆示例，直接复用之前定义的 Agent：

python复制import sqlite3

conn = sqlite3.connect("memory.db")
conn.execute("CREATE TABLE IF NOT EXISTS user_pref(user_id TEXT, key TEXT, value TEXT)")

def save_pref(user_id, key, value):
    conn.execute(
        "INSERT INTO user_pref(user_id, key, value) VALUES (?, ?, ?)",
        (user_id, key, value)
    )
    conn.commit()

def load_pref(user_id):
    rows = conn.execute(
        "SELECT key, value FROM user_pref WHERE user_id=?",
        (user_id,)
    ).fetchall()
    return {k: v for k, v in rows}

在 Agent 每次启动时读取 load_pref，把用户偏好拼进 system prompt；对话结束时把新偏好 save_pref。这种设计在工程上更可控，因为你可以随时修改存储结构和数据范围。

如果偏好内容很多、变化很快，再考虑向量检索。使用 Chroma 作为本地向量库时示例如下：

python复制from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings

vector_store = Chroma(
    embedding_function=OpenAIEmbeddings(),
    persist_directory="./chroma_db"
)

需要说明的是，向量检索适合“模糊召回历史片段”，不适合精确覆盖用户状态。生产里通常把 SQLite/MySQL 的结构化记忆和向量语义记忆一起用。

6. 人机协同：关键操作必须人来确认

真实的业务系统不可能让大模型直接执行所有动作。删除数据、发送邮件、执行命令、批量更新，这些高权限操作需要人工审批。LangGraph 提供的方案是通过 interrupt_before 在工具节点前中断执行流。

python复制agent = create_react_agent(
    model=llm,
    tools=tools,
    checkpointer=memory,
    interrupt_before=["tools"],
)

config = {"configurable": {"thread_id": "user-002"}}

# 第一次 invoke 会停在执行工具前
events = agent.stream(
    {"messages": [("human", "帮我把库存数量改为 0")]},
    config=config,
    stream_mode="values",
)

for event in events:
    # 在这里获取当前状态，并打印给审批人员
    if event.get("messages"):
        print(event["messages"][-1].content)

当流式输出停住后，人工审核工具调用参数。确认无误后，再次使用同一个 config 调用 Agent，它就会从断点继续执行：

python复制agent.invoke(None, config=config)

这是把“人机协同”落进代码的关键模式。你在项目中可以把中断事件推送到企业内部审核平台，比如飞书审批、邮件确认、管理后台按钮。只有审批通过后才调用 agent.invoke(None, config) 放行。

判断流程是否成功：如果模型想要调用工具，但执行停在工具节点前，说明中断生效；审批后能继续拿到工具结果，说明恢复生效。

7. 安全护栏：防止 AI 越权、防提示词注入、限制输出

安全护栏不是单一功能，而是一组工程措施。

7.1 工具白名单

不管模型推理出什么工具调用，最后真正能执行的只有白名单里的工具：

python复制ALLOWED_TOOLS = {"get_weather", "get_humidity"}

def filter_tool_calls(tool_calls):
    return [
        tc for tc in tool_calls
        if tc.get("name") in ALLOWED_TOOLS
    ]

任何不在白名单里的调用直接拦截，不落执行节点。

7.2 参数校验

模型生成的参数不一定合法。比如电商系统里，商品数量不能为负数，金额不能超过某个阈值。在工具函数内增加 pydantic 校验或自定义校验逻辑：

python复制from pydantic import BaseModel, Field

class UpdateInventoryArgs(BaseModel):
    product_id: str = Field(description="商品ID")
    quantity: int = Field(ge=0, description="变更后数量，不能为负数")

def update_inventory(args: dict):
    validated = UpdateInventoryArgs(**args)
    # 执行库存更新
    return {"product_id": validated.product_id, "quantity": validated.quantity}

校验失败就返回错误信息，让模型重新思考，而不是直接执行。

7.3 输出结构校验

如果 Agent 需要对接下游系统，建议强制要求结构化输出，避免模型返回格式漂移。

python复制from typing import TypedDict

class FinalReply(TypedDict):
    reply: str
    tool_used: str
    confidence: float

这样调试时能清楚知道哪些字段是模型生成的、哪些是模板拼出来的，也方便做后续审计。

7.4 提示词注入防护

把网络内容、用户上传文件中的文本拼接进 prompt 时，要按不可信数据处理。比较实用的策略是：

• 不直接执行用户文本中声明的“系统指令”；
• 涉及敏感操作前，强制走人机协同节点；
• 保留工具参数日志，出现异常调用时可追溯原因。

从材料给出的内容来看，完整体系里专门安排安全护栏章节，目的就是提醒开发者把“护栏”当成 Agent 的一等公民，而不是最后打补丁。

8. 实际运行与效果验证

建议按以下测试矩阵逐项验证：

运行脚本时建议加一个打印事件节点的函数。这样每一步的输入输出都能看到：

python复制for event in agent.stream({"messages": [("human", "上海适合跑步吗")]}, config=config, stream_mode="values"):
    msg = event["messages"][-1]
    print(">>>", msg.type, msg.content)

出现问题时先定位是哪一类问题：模型没调用工具、调了工具但参数解析失败、工具执行报错、人机协同没触发、记忆没有恢复。每一类问题对应不同的排查路径。

9. 接口 API 与批量任务设计

9.1 包装成 HTTP 服务

把 Agent 封装成 FastAPI 服务，业务系统就能直接对接：

python复制from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class ChatRequest(BaseModel):
    text: str
    thread_id: str = "default"

@app.post("/chat")
def chat(req: ChatRequest):
    config = {"configurable": {"thread_id": req.thread_id}}
    result = agent.invoke(
        {"messages": [("human", req.text)]},
        config=config,
    )
    return {"reply": result["messages"][-1].content}

启动命令：

bash复制uvicorn main:app --host 127.0.0.1 --port 8000

9.2 批量任务

处理多个用户的批量请求时，关键是控制并发和重试，而不是单方面提高线程数。一个基础版本：

python复制import concurrent.futures

def process_one(text, thread_id):
    try:
        response = agent.invoke(
            {"messages": [("human", text)]},
            config={"configurable": {"thread_id": thread_id}},
        )
        return response["messages"][-1].content
    except Exception as e:
        return {"error": str(e)}

with concurrent.futures.ThreadPoolExecutor(max_workers=4) as pool:
    futures = [
        pool.submit(process_one, item["text"], item["thread_id"])
        for item in task_list
    ]
    for f in concurrent.futures.as_completed(futures):
        print(f.result())

批量任务要额外设计：失败重试、超时熔断、结果落库。任何一步失败，都应该能重新拉起而不影响其他任务。

10. 资源占用与性能观察

这个体系的资源消耗主要不在 Python 框架本身，而在模型调用次数和提示词长度。

10.1 模型调用次数

简单 Agent 每轮可能只调用一次模型。加了工具、记忆、审批之后，模型调用次数会明显上升。每一次工具返回后，模型可能需要再生成一次总结。建议在日志里记录每一轮模型调用次数，避免工具链过长导致延迟过高。

10.2 Token 消耗

短期记忆越长，每次请求消耗的 token 越多。这个体系的记忆策略要做取舍：

• 不是把全部历史消息永远塞进上下文；
• 可以只保留最近 N 轮完整消息，再附一条历史摘要；
• 长期记忆只注入与当前任务相关的结构化数据。

10.3 本地模型与显存

如果使用云端模型，服务端显存与本机无关。如果使用 Ollama 之类的本地模型，显存占用由模型参数规模、量化精度、上下文长度决定。设备测试前建议先跑一个最小模型确认推理链路，再逐步加大到目标模型。上下文越长，综合开销越大，实际占用需要按本机模型版本测试。

10.4 监控指标

建议至少记录以下几点：

• 每次模型调用的延迟
• 工具调用成功率和失败原因
• 当前会话消息数
• 每轮 task 的总耗时

这些指标能快速暴露是模型本身慢、工具接口慢、还是记忆读取慢。

11. 常见问题与排查方法

12. 最佳实践与合规边界

12.1 工程建议

• Agent 脚本要分成独立模块：模型封装、工具定义、记忆存储、图状态、HTTP 路由。不要把所有代码写在一个文件里。
• 生产环境为每个用户、每个业务线单独设置模型调用配额。
• 工具执行必须有 trace 日志，保留 request id、工具名称、参数、结果、耗时，方便审计和回放。
• 人机协同不该只依赖模型自觉判断，而应从系统层面限制高风险工具的执行权限。
• 首次上线时用小流量灰度，Agent 回答全部进入人工复核队列，确认无误后再扩大范围。

12.2 安全合规边界

涉及用户数据、隐私信息、人脸声音素材、版权内容时，必须遵守业务线和平台的双重授权要求，不把未经处理的敏感数据直接传入第三方模型。涉及金融、医疗、审批等场景时，关键操作应保留完整决策链路记录，人工确认步骤不能省略。模型输出的内容在对外发布前也要做复核，避免因模型幻觉产生误导。

13. 总结与下一步

LangChain 新版完整体系最有价值的地方，是把 Agent 从一个“调函数”的小玩具变成了可插拔节点的业务流程。工具调用负责能力边界，长短记忆负责上下文延续，人机协同负责风险控制，安全护栏负责防越权和防注入。四者结合，才是工程化 Agent 的完整形态。

最容易踩的坑集中在两点：一是记忆只在内存里，没做持久化，进程一重启就“失忆”；二是安全护栏放在最后才设计，导致模型已经能触达敏感工具。先跑通第 3 到第 6 节的最小示例，再加 HTTP 服务，再加批量任务，逐步积累评估数据。

下一步可以继续扩展的方向包括：向量库接入对话历史、审批事件推送到企业 IM 系统、多 Agent 协作、以及更完整的内容安全审计。建议先把这套最小体系跑起来，再根据线上日志逐步加功能。