LangChain Agent完整指南：工具调用、记忆、人机协同与安全护栏

在实际项目里，LangChain 的用法往往只停留在“给模型一个 System Prompt，再挂两个工具函数”。一旦遇到真实业务，比如需要让 Agent 查询订单、修改备注、在关键操作前征求人工确认、同时记住用户刚才说过的话，就会快速暴露出三个问题：工具调用到底是怎么被选中的、记忆到底存在哪里、安全边界由谁来兜底。这篇博客会围绕新版 LangChain 完整的编排思路，从工具调用、长短期记忆、人机协同、安全护栏四个维度展开，最后用一个小型客服 Agent 把四者串起来。读者可以跟着代码逐步搭建、验证和排查，最终能够结合自身项目设计一套可落地的 Agent 体系。

1. 先理解 LangChain Agent 体系中的核心角色：模型、工具、记忆、控制流

很多初学者把 Agent 理解成“模型加工具”，这是不完整的。一个能进生产的 Agent 需要同时具备四个组成部分：模型负责理解意图和生成文本，工具负责执行具体操作，记忆负责保存和检索上下文，控制流负责决定调用顺序、是否停止、是否请求人工介入。LangChain 的价值并不是“封装了模型 API”，而是把这四部分组织成一个可以观察、可以修改、可以回滚的编排过程。

1.1 不只是“接一个 LLM”，LangChain 解决的是编排问题

直接调用模型 API 时，你写的是一次性的“请求-响应”。例如向模型发送一个问题，得到一段答案。如果这个过程中需要查数据库、调接口、读取文件，模型本身无法完成，它只能告诉你“需要调用一个工具”，然后你就得自己写代码去调用，再把结果拼回新的 Prompt 发给模型。

LangChain 把这一过程抽象成 Agent 循环：模型先判断是否需要工具，如果需要，则返回工具名和参数；框架负责执行工具；执行结果重新回到模型，模型再决定继续调用还是给出最终回答。这个循环的核心是“控制流”。新版 LangChain 更推荐使用 LangGraph 这类底层编排工具来替换早期的 AgentExecutor，因为 LangGraph 允许你把循环节点、条件边、人工中断、并行节点显式画出来，开发时能明确看到 Agent 在哪个环节卡住。

1.2 工具调用是 Agent 能力的边界，而不是模型能力的边界

模型本身不能查询实时库存，也不能修改数据库。Tool（工具）的定义就是把这类外部能力注册给模型，让模型知道什么时候可以用。工具的描述和参数模式会成为模型 Prompt 的一部分，模型根据当前用户意图、对话历史、工具描述，来决定是否调用以及传入什么参数。

因此工具设计直接决定 Agent 的能力边界。你的工具列表里没有“退款”工具，模型再聪明也不会自动退款；工具描述含糊，模型就容易传错参数。把工具理解成“Agent 的外部能力目录”，比把它理解成“函数集合”更准确。

1.3 记忆、协同、安全在体系中的位置

记忆解决的是“上下文从哪里来”的问题。如果每次请求都是独立发送，用户说“把刚才那个订单取消”，模型不知道“刚才那个”是哪个订单。短期记忆通过消息列表保存本次会话，长期记忆通过向量库保存历史事实和用户偏好。

人机协同解决的是“哪些操作不能让模型自主决定”的问题。比如删除数据、发送大批量通知、修改高额订单，这类操作需要插入人工确认节点，模型执行到该节点时暂停并等待用户审批。

安全护栏解决的是“出错时损失最小”的问题。包括工具白名单、参数校验、输出过滤、权限控制、日志审计。这四个部分不是独立模块，而是在控制流中互相配合：记忆提供上下文，工具扩展能力，人机协同控制风险，安全护栏兜底。

2. 工具调用：让 Agent 能真正操作外部系统

工具调用是本体系的第一块基石。这里从工具定义、注册、解析、异常处理四个环节展开。

2.1 定义工具的方式和输入输出规范

在 LangChain 中定义工具最常用的是 @tool 装饰器。下方示例定义了一个查询订单状态的工具，注意函数名和 docstring 要写清楚，因为它们会直接暴露给模型。

python复制from langchain.tools import tool

@tool
def get_order_status(order_id: str) -> str:
    """查询订单状态。仅用于查询，不要修改任何数据。
    参数 order_id 是订单编号，格式为 ORD 开头加数字。
    """
    # 这里替换成你的真实后端 API 调用
    result = f"订单 {order_id} 当前状态为已发货"
    return result

如果工具参数比较复杂，推荐使用 StructuredTool 并明确声明参数模型：

python复制from pydantic import BaseModel, Field
from langchain.tools import StructuredTool

class CancelOrderInput(BaseModel):
    order_id: str = Field(description="订单编号，ORD开头")
    reason: str = Field(description="取消原因")

def cancel_order(order_id: str, reason: str) -> str:
    # 调用取消订单服务
    return f"订单 {order_id} 已取消，原因: {reason}"

cancel_order_tool = StructuredTool.from_function(
    func=cancel_order,
    name="cancel_order",
    description="取消指定订单，需要明确取消原因。这是一个高风险操作。",
    args_schema=CancelOrderInput,
)

工具描述的几个原则：

• 描述里明确写“可以做什么、不可以做什么、参数格式是什么”。
• 判定工具是否可用的词要清楚，比如“仅用于查询”和“用于修改”。
• 不要在一个工具里塞太多职责，遵循单一职责，模型调用准确率会更高。

2.2 注册工具给 Agent，并让模型学会选择工具

工具定义好后，需要绑定到 Agent 的模型上。新版 LangChain 推荐使用 create_tool_calling_agent 结合 LangGraph，也可以用 bind_tools 把工具列表绑定给 ChatModel。

python复制from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent
from langgraph.prebuilt import create_agent

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

# 使用 LangGraph 的 create_agent 快速构建
agent = create_agent(llm, tools)

注意，模型选择工具的前提是：

• 模型支持 function calling 或 tool calling。
• 工具的数量不要一次绑定太多，超过 20 个之后选择准确率会明显下降。
• 工具描述过长会挤占 Prompt 窗口，需要做精简。

2.3 工具调用的结果解析和异常处理

工具调用不是“调用完就结束”，结果要回到模型再做判断。如果工具出错，比如接口超时、订单不存在、参数不合法，模型需要收到明确的错误反馈，然后生成对用户友好的说明或尝试其他工具。

业务代码里建议对每个工具做异常封装：

python复制from langchain.tools import tool

@tool
def get_order_status(order_id: str) -> str:
    """查询订单状态。"""
    try:
        # 调用真实订单服务
        status = query_order_api(order_id)
        if status is None:
            return "未查询到该订单，请确认订单号是否正确。"
        return f"订单状态: {status}"
    except Exception as e:
        return f"查询失败: {str(e)}。请提醒用户稍后重试。"

这样模型收到的是“可理解的失败信息”，而不是堆栈异常。关键坑在于：不要返回未经处理的原始异常，模型会误解为业务结果，甚至可能尝试用错误信息去回答问题。

3. 长短期记忆：Agent 如何记住上下文、事实和用户偏好

记忆是 Agent 与“无状态单次调用”最明显的区别。记忆设计不好，会出现“聊十轮就忘记用户名字”的情况。

3.1 短期记忆：会话窗口内的消息管理

短期记忆通常由一个消息列表组成，包含 System、User、Assistant、Tool 等消息类型。LangChain 中用 MessagesPlaceholder 或 ChatMessageHistory 管理。

python复制from langchain.memory import ChatMessageHistory

history = ChatMessageHistory()
history.add_user_message("我的订单号是 ORD123456")
history.add_ai_message("好的，订单 ORD123456 已记录。")

在实际应用中，短期记忆要注意上下文窗口限制。简单把所有历史塞给模型，一旦消息量大就会超限。建议做摘要式记忆，既能保留关键信息，又能控制长度：

python复制from langchain.memory.summary import ConversationSummaryMemory

memory = ConversationSummaryMemory(llm=llm)
memory.save_context({"input": "我想查订单"}, {"output": "好的，请提供订单号"})

3.2 长期记忆：向量库存储和检索

长期记忆的目的是跨会话记住事实，比如用户的联系方式、偏好、历史订单。LangChain 中常用 VectorStoreRetrieverMemory，它把对话历史转化为向量存入向量库，需要时按语义相似度取回相关片段。

python复制from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain.memory import VectorStoreRetrieverMemory

vectorstore = Chroma(embedding_function=OpenAIEmbeddings())
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
memory = VectorStoreRetrieverMemory(
    retriever=retriever,
    memory_key="chat_history",
    return_messages=True,
)

使用时，在 Agent 的 Prompt 中留出占位符：

python复制from langchain.prompts import PromptTemplate

prompt = PromptTemplate.from_template(
    """你是一个客服助手。以下是和当前用户相关的历史记忆:
{chat_history}

当前用户输入:
{input}"""
)

长期记忆需要注意检索粒度。不要存整段无意义的闲聊，而是把“事实”提取出来再存储，比如“用户喜欢顺丰快递”“用户上次退单原因是地址错误”。否则向量检索会召回大量无关信息，反而干扰模型判断。

3.3 长短期记忆的组合策略与更新机制

比较稳妥的组合策略是三层：

• 当前窗口内的最近几轮对话，原样拼入 Prompt，保证实时性。
• 整个会话的摘要放在短期记忆区，压缩旧内容。
• 跨会话的用户事实存入长期向量库，按需检索。

更新机制要明确：什么信息值得写入长期记忆，什么时候更新。建议用单独的“记忆提取 Agent”在每次会话结束时，从对话中抽取结构化事实，写入向量库。写入前要做去重和覆盖，避免同一个用户的多个版本互相冲突。

下表对比不同记忆方案的使用场景：

在实际项目中，不要把全部记忆都塞进一个 Prompt。先看模型上下文窗口上限，再做分级压缩。

4. 人机协同：在 Agent 自主执行和人工介入之间找到平衡

不是所有操作都适合让模型独立完成。人机协同（Human-in-the-loop）是指在某些高风险节点主动暂停 Agent 执行，等待人工审批后再继续。

4.1 为什么需要人在回路，而不是全程自动

模型对语言的判断可能有误，工具调用的参数也可能传错。比如“取消订单”，模型认为用户权限足够，但实际是管理员才能操作；再比如批量发送邮件，模型可能因为 Prompt 注入而误触发送。人机协同的价值不是降低效率，而是在“不可逆操作”和“高影响操作”前增加一道审批闸门。

可自动执行的操作：查询类、只读类、低风险计算类。

需要人工确认的操作：删除、写入、转账、发送消息、批量操作、涉及第三方系统的副作用操作。

4.2 用中断和审核节点实现人工确认

在 LangGraph 中，可以在节点之间加入 interrupt 或自定义审核节点。以取消订单为例，Agent 先去调用预取消校验工具，然后进入人工审核节点，等待操作员确认后才执行真正的取消。

python复制from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated

class AgentState(TypedDict):
    order_id: str
    need_confirm: bool
    confirmed: bool

async def precheck(state: AgentState) -> AgentState:
    # 判断是否为高风险操作
    return {"need_confirm": True}

async def require_approval(state: AgentState) -> AgentState:
    # 这里在实际系统中会阻塞等待审批结果
    if not state.get("confirmed"):
        return {"confirmed": False}
    return {"confirmed": True}

async def execute_cancel(state: AgentState) -> AgentState:
    # 真正执行取消
    return {"result": "已取消"}

builder = StateGraph(AgentState)
builder.add_node("precheck", precheck)
builder.add_node("approval", require_approval)
builder.add_node("execute_cancel", execute_cancel)
builder.add_edge("precheck", "approval")
builder.add_edge("approval", "execute_cancel")
builder.add_edge("execute_cancel", END)

上面是流程示意。生产环境中，require_approval 节点不应直接返回，而是把状态写入待审批表，推送通知给操作员，操作员通过接口审批后触发流程恢复。

4.3 协同流程的设计：审批、暂存、恢复

设计人机协同流程时，至少要考虑三件事：

• 审批超时如何处理。操作员长时间未审批，是默认拒绝还是自动取消，需要业务规则确定。
• 流程如何暂存和恢复。Agent 的中间状态要持久化，不能只放在内存里，否则服务重启后审批结果丢失。
• 审批人操作界面。常见做法是提供一个小管理后台，展示 Agent 计划执行的工具、参数和理由，操作员可以“通过”“拒绝”“修改参数后通过”。

一个容易被忽略的坑是：模型在等待审批时，可能会继续生成文本，造成“回答已经发送给用户但操作还没执行”的错觉。因此人机协同节点要明确暂停整个流程，在审批通过前不输出任何最终结果。

5. 安全护栏：防止 Agent 越权、误操作和输出风险

安全护栏是生产级 Agent 和 Demo 级 Agent 的最大区别。没有护栏，模型越权调用只是概率问题。

5.1 输出侧护栏：格式校验、敏感词、内容过滤

模型输出层需要做两类校验：

第一，格式校验。如果 Agent 需要输出 JSON，必须在代码层面用 Pydantic 解析，解析失败则触发重试或人工处理，不能直接把模型文本返回给下游。

python复制from pydantic import BaseModel, ValidationError

class Reply(BaseModel):
    content: str
    tool_used: str | None = None

try:
    parsed = Reply.model_validate_json(raw_output)
except ValidationError:
    # 触发重试或安全拒绝
    parsed = Reply(content="系统无法生成合规响应，请稍后重试。", tool_used=None)

第二，内容过滤。对输出的文本做敏感词检测、PII 检测（身份证号、手机号、银行卡号）。可以在输出前套一层过滤器，命中敏感规则则替换为脱敏内容。

5.2 操作侧护栏：工具白名单、权限控制、成本限制

模型可以调用工具，但不代表可以调用所有工具。实际项目中需要建立“工具白名单”和“操作权限映射”。

python复制ALLOWED_TOOLS = {
    "user": ["get_order_status", "get_user_info"],
    "operator": ["get_order_status", "cancel_order", "refund_order"],
}

def check_tool_permission(user_role: str, tool_name: str) -> bool:
    return tool_name in ALLOWED_TOOLS.get(user_role, set())

在 Agent 执行链中添加权限检查节点，工具执行前先校验当前用户角色是否允许调用该工具。同时还要加成本限制，例如每轮对话单用户成本上限、单次任务工具调用次数上限、模型 Token 上限，防止恶意输入导致巨额费用。LangGraph 中可以设置 recursion_limit，或者自定义计数器节点来统计调用次数。

5.3 审计与日志：可回放、可追溯

所有 Agent 行为都应该落日志，至少包含：

• 用户输入原文。
• 模型调用的完整 Prompt 或消息摘要。
• 每次工具调用名称、参数、返回值和耗时。
• 是否经过人工审批，审批人和审批结论。
• 最终输出内容。

日志的关键不是记录，而是支持回放。当线上发生误操作时，可以通过日志恢复完整的 Agent 执行链路，找到是哪一步出了问题，是模型选错工具，还是工具内部逻辑错误。建议按 RequestId 串联所有日志，方便检索。

注意：不要只记录最终结果，中间的工具调用过程同样重要。排查“为什么取消订单”这个问题时，没有中间日志几乎无从下手。

6. 一个综合示例：带记忆、工具、确认和安全检查的客服 Agent

把前四章的内容落成一个可运行的综合示例。这里以“客服 Agent 帮用户查询订单并要求取消订单”为例。

6.1 场景和需求

用户通过客服对话入口说“查一下 ORD123456 的状态”，然后说“如果已经发货就不要取消，如果还在待发货就帮我取消”。Agent 需要做四件事：

• 查询订单状态（只读工具）。
• 根据状态判断是否执行取消。
• 如果执行取消，则需要操作员确认。
• 整个流程记录到日志，输出给用户可理解的回答。

6.2 代码结构

文件建议按以下结构组织：

text复制customer_agent/
├── agent.py
├── tools.py
├── memory.py
├── security.py
├── graph.py
└── main.py

tools.py 中定义查询和取消工具：

python复制from langchain.tools import tool

@tool
def query_order_status(order_id: str) -> str:
    """查询订单状态。只读操作。"""
    status_map = {"ORD123456": "待发货", "ORD654321": "已发货"}
    return status_map.get(order_id, "订单不存在")

@tool
def execute_cancel(order_id: str) -> str:
    """执行订单取消。这个工具会真正修改订单状态，需要审批后才可调用。"""
    return f"订单 {order_id} 已取消"

security.py 中实现权限检查：

python复制from enum import Enum

class Role(str, Enum):
    user = "user"
    operator = "operator"

def check_tool_access(role: Role, tool_name: str) -> bool:
    if role == Role.user:
        return tool_name == "query_order_status"
    if role == Role.operator:
        return tool_name in {"query_order_status", "execute_cancel"}
    return False

graph.py 中构建流程：

python复制from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from typing import Literal

class State(dict):
    order_id: str = ""
    status: str = ""
    role: Role = Role.user
    tool_approved: bool = False
    result: str = ""

def query_node(state: State) -> State:
    state["status"] = query_order_status.invoke({"order_id": state["order_id"]})
    return state

def decide_cancel(state: State) -> Literal["ask_human", "finish"]:
    if "待发货" in state["status"]:
        return "ask_human"
    return "finish"

def check_permission(state: State) -> State:
    if check_tool_access(state["role"], "execute_cancel"):
        state["tool_approved"] = True
    return state

def human_confirm(state: State) -> State:
    # 实际项目中这里会写入审批记录，等待外部接口批准
    # 这里简化：如果已进入该节点，则模拟批准
    state["tool_approved"] = True
    return state

def execute_node(state: State) -> State:
    if state["tool_approved"]:
        state["result"] = execute_cancel.invoke({"order_id": state["order_id"]})
    else:
        state["result"] = "没有权限取消订单"
    return state

graph = StateGraph(State)
graph.add_node("query_node", query_node)
graph.add_node("check_permission", check_permission)
graph.add_node("human_confirm", human_confirm)
graph.add_node("execute_node", execute_node)
graph.add_conditional_edges("query_node", decide_cancel, {"ask_human": "check_permission", "finish": END})
graph.add_edge("check_permission", "human_confirm")
graph.add_edge("human_confirm", "execute_node")
graph.add_edge("execute_node", END)

app = graph.compile(checkpointer=MemorySaver())

调用方式如下：

python复制from langchain_core.runnables import RunnableConfig

config = RunnableConfig(recursion_limit=10)
initial_state = State(order_id="ORD123456", role=Role.user)
result = app.invoke(initial_state, config)
print(result.get("result"))

6.3 运行验证

正常验证路径：

• 输入订单 ORD123456，状态为“待发货”，走到人工确认节点，模拟审批后执行取消。
• 输入订单 ORD654321，状态为“已发货”，直接结束，不执行取消。

这个示例展示了记忆、工具、人机协同和安全检查如何组合。实际项目中，你还需要把 state 中的 role 替换成真实的会话用户身份，把 human_confirm 替换成异步审批接口，并把日志记录到统一日志平台。

7. 常见问题排查清单及最佳实践

搭建 Agent 体系的过程中，最常见的几类问题需要明确排查路径。

7.1 工具循环卡死、超时和误调用排查

现象：

• Agent 反复调用同一个工具，迟迟不给最终回答。
• Agent 调用工具时参数明显错误，比如把字母 O 当成数字 0。
• 工具执行时间长导致整个请求超时。

排查步骤：

1. 查看 Agent 的中间日志，确认模型每一次选择工具后的输出。
2. 检查工具描述是否足够明确。参数格式是否写清示例。
3. 检查工具绑定数量是否过多，工具列表超过 15 到 20 个时会明显影响选择准确率。
4. 检查模型类型是否支持 tool calling，以及是否是旧版模型。
5. 检查有没有设置 recursion_limit 或循环最大步数。

解决方案：

7.2 记忆混乱和检索不准确排查

现象：

• 用户明确说过的事实，Agent 回答时却忘记了。
• 向量检索返回的“历史记忆”与当前问题无关，反而干扰回答。

排查步骤：

1. 检查短期记忆是否真的保存了消息，确认内存存储或 Redis 配置正确。
2. 检查摘要记忆是否漏掉关键事实，摘要模型是否用了较弱的模型。
3. 检查向量库是否存入了噪声对话，比如把“你好”“再见”也算作记忆。
4. 检查检索阈值，是否取回过多或过少片段。

最佳做法是先把记忆分为“用户事实”和“对话过程”。用户事实用结构化字段存数据库或向量库，对话过程用摘要或消息列表。不要把所有消息一股脑写入向量库。

7.3 人机协同流程卡住怎么办

现象：

• Agent 执行到审批节点后没有继续响应。
• 用户看到机器人说“正在取消订单”，但订单一直没有变化。

排查步骤：

1. 检查审批等待是否使用了异步任务或消息队列，确认审批回调能否触发流程恢复。
2. 检查流程状态是否持久化，使用 MemorySaver 只在进程内有效，支持断点恢复需要使用数据库持久化。
3. 检查审批接口是否被权限拦截，操作员点击通过后，回调 Body 是否携带正确的 thread_id。

人机协同最核心的设计原则是“中间状态可恢复”。不要把流程状态只保存在内存里，否则进程重启后就无法恢复审批结果。

7.4 安全护栏失效的常见原因

现象：

• 模型输出了敏感信息。
• 用户通过 Prompt 注入让 Agent 调用了未授权的工具。

排查步骤：

1. 检查输出侧是否只做了模型输出后过滤。如果过滤在生成前没有做，模型依然可能产生敏感内容，因此输出过滤不能省。
2. 检查工具调用是否每次都在执行前做了权限校验。只在页面层控制工具列表但执行层不校验，等于没有护栏。
3. 检查日志是否完整记录工具参数，如果没有，事后无法追溯误操作。

推荐在生产环境中增加一层“人类可读审计摘要”，每次 Agent 执行完，自动生成一条中文摘要，写明“用户说了什么、Agent 调用了什么工具、结果是什么、是否审批”。这比只看原始日志更容易定位问题。

7.5 面向生产的发布前检查清单

在把 Agent 体系发布到线上之前，可以按下面清单逐项确认：

• 工具列表是否有版本号，工具签名变更后是否通知了模型。
• 是否设置最大调用轮数、单用户成本上限、单次请求超时。
• 是否记录完整执行链路日志，并且日志能按用户 ID 或 RequestId 检索。
• 是否对模型输出做了格式校验和内容过滤。
• 是否对高风险工具增加人工审批节点。
• 审批状态是否持久化，服务重启后能否恢复。
• 是否做了 Prompt 注入防护，例如不在工具参数中拼接不可信指令。
• 是否限制了向量库检索返回数量，避免超大上下文。
• 是否做了容量测试，工具调用并发高时是否拖垮下游服务。

8. 从简单 Demo 到完整体系的下一步

这篇博客把 LangChain 的 Agent 从“模型加工具”扩展成“工具 + 记忆 + 人机协同 + 安全护栏”的完整体系。四个能力不是可选项，而是进入真实业务时绕不开的要求。工具调用要能让模型稳定选择并正确传参；记忆要能区分短期上下文和长期事实；人机协同要能对高风险操作插入审批节点；安全护栏要能覆盖输出、操作和审计三个层面。

建议读者从一个小场景开始练习，比如给客服 Agent 增加一个查询工具和一个人工审批节点，先跑通流程，再逐步加入向量记忆和成本限制。不要一开始就追求大而全，否则会把 Agent 的不可靠性和业务复杂度混在一起，排查起来非常痛苦。

下一步可以继续研究 LangGraph 的流式输出、多 Agent 协作、子图复用，以及如何把内存检查点换成外部数据库存储。只有把执行过程拆成可观察、可暂停、可恢复的图结构，Agent 才能真正用于生产环境。