这次我们来看 LangChain Agent 的底层运行方式。很多教程提到 Agent 时只会说一句“模型可以调用工具”但实际跑起来之后你会发现整个过程是一个循环推理Thought- 行动Action- 拿到反馈Observation- 再推理……直到模型认为可以给出最终答案。这个循环就是 ReAct 模式的核心。为了讲清楚这个机制本文不打算用一个现成的复杂 Agent 案例而是自己写一个工具甚至故意把工具设计成“不可用”——要么执行时抛异常要么工具描述写得与功能完全不一致。用这种反例来演示 Agent 底层的推理-行动-反馈循环反而能看清最真实的行为LLM 拿到 Observation 后怎么想、Agent 处理工具错误时会不会崩溃、循环在什么情况下会终止。这篇文章会覆盖LangChain Agent 核心能力速览、适用场景与使用边界、ReAct 循环原理、环境准备、自定义工具代码、AgentExecutor 的运行日志解读、接口 API 封装、批量任务、资源占用与排查清单。适合刚学 LangChain、想搞清楚 agent 开发底层逻辑的读者如果你已经在用 LangGraph 或 MCP这篇文章也能帮你理解为什么工具描述和工具返回值格式对 Agent 行为影响这么大。1. LangChain Agent 核心能力速览能力项说明项目类型Agent 开发框架 / 大模型编排层开源情况LangChain 开源项目社区使用量很大核心机制ReAct 模式推理-行动-反馈循环自定义工具支持通过tool装饰器或Tool类定义 Python 工具支持模型OpenAI、Anthropic、Ollama 本地模型等具体取决于 LangChain 适配器硬件消耗Agent 编排层本身不吃显存显存/内存取决于所接的大模型推理服务启动方式Python 脚本开发可封装为 Web API 服务是否支持 API可以FastAPI 封装后对外提供 HTTP 接口是否支持批量任务可以用循环或线程池执行多次invoke需自行管理任务队列适合场景客服问答、信息查询、数据分析、自动化流程、多工具调度先说结论LangChain Agent 的学习重点是两个部分一个是工具的定义方式另一个是执行器对循环的控制。理解了这两个部分后面再学 LangGraph、MCP 工具接入都会轻松很多。2. 适用场景与使用边界LangChain Agent 适合需要“模型自主决定调用哪个工具”的场景。典型例子是用户问“现在几点了”时模型先判断这个问题自己无法直接回答然后调用时间工具用户问“北京天气如何”时模型调用天气工具用户直接问“11 等于几”时模型可以不走工具直接输出答案。这种“根据问题动态选择工具”的能力是 Agent 相比普通 Prompt 问答的最大区别。Agent 不适合的场景也要说清楚。如果只是简单的固定流程比如“先做 A 再做 B 再做 C”直接用代码写流程更稳定成本也更低。Agent 每次推理都需要调用大模型轮次越多 token 消耗越大响应时间越长。尤其在高并发业务里无限制的 ReAct 循环会带来不可控的延迟和成本。合规边界方面本文用到的工具都是本地 Python 函数不存在第三方数据泄露问题。但如果你要接入真实业务系统需要确认接口调用权限如果要处理用户数据先做脱敏如果要对接外部大模型 API注意控制成本并遵守对应服务商的使用条款不要拿未经授权的账号和高并发任务去压测。3. LangChain Agent 底层运行机制推理-行动-反馈循环Agent 的底层原理可以拆成四步循环执行模型收到用户问题先做推理Thought判断下一步怎么做。模型输出行动指令Action和行动参数Action Input指定调用哪个工具。执行器根据 Action 找到并执行工具得到工具返回结果Observation。模型把 Observation 拼回上下文再次做推理。如果认为已经足够就输出 Final Answer否则继续输出新的 Action。这个循环由AgentExecutor驱动。AgentExecutor内部维护一个agent_scratchpad每次工具返回结果后会把“Action Observation”追加到 scratchpad 中再传给模型继续推理。模型不是一次调用就给出最终答案的而是多轮对话式的自我更新。打开 verbose 日志后可以看到类似这样的执行过程这里给出一个预期输出示例具体文本取决于你使用的模型 Entering new AgentExecutor chain... Thought: 用户想知道当前时间我可以使用 get_current_time 工具。 Action: get_current_time Action Input: {} Observation: Tool get_current_time failed to execute: time service unavailable Thought: 工具执行失败了可能是时间服务暂时不可用我再尝试一次。 Action: get_current_time Action Input: {} Observation: Tool get_current_time failed to execute: time service unavailable Thought: 两次尝试都失败了我无法获取可靠时间应该告知用户当前无法获取时间。 Final Answer: 抱歉我暂时无法获取当前时间。 Finished chain.这个日志就是理解 Agent 最好的入口。可以看到工具报错并不会让整个 Agent 崩溃Observation 变成了一个错误文本模型会基于这个文本重新推理。这也是 ReAct 循环“鲁棒性”的来源。4. 环境准备与依赖安装先把运行环境准备好。建议使用 Python 3.10 及以上版本并创建独立的虚拟环境避免和系统 Python 环境互相污染。python -m venv .venv source .venv/bin/activate安装 LangChain 相关依赖。本文重点演示 AgentExecutor 和 ReAct依赖包括langchain、langchain-core、langchain-community以及模型适配器。pip install langchain0.2 langchain-core langchain-community模型部分有两种选择。第一种是调用 OpenAI 兼容接口需要准备 API Key第二种是本地模型例如 Ollama不需要额外 API Key但本地推理速度取决于机器性能。# 如果使用 OpenAI 兼容接口 pip install langchain-openai # 如果使用 Ollama 本地模型 pip install langchain-ollama其中 OpenAI 兼容接口的初始化写法如下代码中的base_url需要按你的服务商地址修改实际项目中建议通过环境变量传入 API Key不要硬编码在源码里。from langchain_openai import ChatOpenAI llm ChatOpenAI( modelgpt-4o-mini, temperature0, api_keyyour-api-key, base_urlhttps://api.openai.com/v1, )如果你的环境里已经装好了 Ollama并且本地拉取过模型可以这样初始化。这里用qwen2.5:7b作为示例模型名具体名称以ollama list的输出为准。from langchain_ollama import ChatOllama llm ChatOllama( modelqwen2.5:7b, temperature0, )首次运行前可以用一个最简的 Prompt 验证模型连通性再继续接下来的 Agent 实验。5. 自定义“不可用工具”演示 Agent 底层循环这一节是全文的核心。我们故意写一个不可用工具让 Agent 在执行过程中不断拿到错误 Observation从而观察 ReAct 循环的行为。5.1 场景 A工具实现抛异常先定义一个工具它的功能描述是“获取当前系统时间”但实现里直接抛异常模拟底层时间服务不可用的情况。import datetime from langchain.agents import AgentExecutor, create_react_agent from langchain_core.prompts import PromptTemplate from langchain_core.tools import tool tool def get_current_time() - str: 获取当前系统时间返回格式 yyyy-MM-dd HH:mm:ss。 raise RuntimeError(time service unavailable, cannot read time source)这里的关键点在于工具名是get_current_time说明里也写了“获取系统时间”但实现故意失败。然后构造 ReAct Agent 和执行器。react_prompt PromptTemplate.from_template(You are a helpful assistant. Use the tools to answer the question if needed. You have access to the following tools: {tools} Use the following format strictly: Question: the input question you must answer Thought: you should always think about what to do Action: the action to take, should be one of [{tool_names}] Action Input: the input to the action Observation: the result of the action ... (this Thought/Action/Action Input/Observation can repeat N times) Thought: I now know the final answer Final Answer: the final answer to the original input question Begin! Question: {input} Thought: {agent_scratchpad}) tools [get_current_time] agent create_react_agent(llmllm, toolstools, promptreact_prompt) executor AgentExecutor( agentagent, toolstools, verboseTrue, max_iterations5, return_intermediate_stepsTrue, ) result executor.invoke({input: 现在几点了}) print(result.get(output))运行之后verbose 日志里会出现多次 Thought、Action、Observation 的循环。模型第一次推理时认为需要调用时间工具工具抛出异常后Observation 是异常信息。模型看到错误后可能再试一次也可能直接告诉用户“暂时无法获取时间”。这个实验说明了一个关键事实工具执行失败不等于 Agent 运行失败。异常被捕获后变成了模型上下文中一段文本模型会基于这段文本继续推理。如果模型足够聪明它会放弃该工具并给出兜底答复如果模型不够强它可能会反复调用同一个失败工具直到max_iterations触顶。5.2 场景 B工具描述与功能不一致第二个不可用场景更隐蔽工具本身能正常执行但描述写错了。这里定义一个工具实际上返回的是当前时间描述却写成“获取指定城市的天气情况”。tool def get_current_time() - str: 获取指定城市的天气情况返回温度、天气状态。 return f{datetime.datetime.now():%Y-%m-%d %H:%M:%S}让 Agent 回答“北京今天天气怎么样”时模型会认为必须调用get_current_time工具因为描述里写了“获取指定城市的天气”。结果工具返回的是一串看起来完全不相关的时间字符串。模型拿到这个 Observation 后会试图强行解释它最终给出一段不可靠的结论。这个实验解释了 Agent 开发中非常重要的一条原则工具描述就是 Agent 的说明书。工具逻辑写得再好描述不准确Agent 的推理就已经被带偏了。反过来说很多 Agent“乱调用工具”的问题根因并不在模型而在工具描述。5.3 场景 C改成正常工具做对比把工具改成正常实现再跑一次同样的问题观察行为差异。tool def get_current_time() - str: 获取当前系统时间返回格式 yyyy-MM-dd HH:mm:ss。 return f{datetime.datetime.now():%Y-%m-%d %H:%M:%S}正常模式下Agent 通常一轮就能完成推理到需要时间工具 - 调用 - 拿到 Observation - 输出 Final Answer。对比 5.1 和 5.2 的场景你会发现模型在不可用工具上消耗的轮次明显更多耗时和 token 消耗也更高。这说明工具质量直接影响 Agent 的效率。5.4 通过 intermediate_steps 查看循环过程如果不想只在日志上看可以把中间步骤保存下来。执行器初始化时设置了return_intermediate_stepsTrue运行结果里会多一个intermediate_steps字段里面记录了每一步的 Action 和 Observation。result executor.invoke({input: 现在几点了}) for step in result.get(intermediate_steps, []): action, observation step print(工具名:, action.tool) print(输入:, action.tool_input) print(反馈:, observation) print(---)这段代码可以用来做日志审计哪个工具被调用了、输入参数是什么、返回结果是什么一目了然。生产环境中把这些步骤写进日志对排查 Agent 行为非常有帮助。6. 功能测试与效果验证写 Agent 代码不是跑通一次就完事的建议按下面几个维度逐项验证。6.1 验证工具是否被正确注册Agent 能识别哪些工具取决于tools列表。如果列表里没有工具Agent 即使看到了描述也无法调用。验证方法是打印工具信息for t in tools: print(t.name, -, t.description)如果工具名或描述不符合预期先在这里修正再跑 Agent。6.2 验证 ReAct 循环是否完整结束执行结果里如果出现output字段说明 Agent 最终输出了 Final Answer。如果执行抛错或者只返回中间步骤没有最终输出常见原因包括模型输出格式不对、max_iterations太低、工具描述和工具实现不一致。判断标准执行成功后结果字典中output非空。6.3 验证故障恢复用 5.1 中抛异常的工具跑几次观察 Agent 是否会在工具失败后给出合理回答。如果 Agent 连续多轮调用同一个失败工具说明模型对 Observation 的利用能力较弱。这时可以调整工具实现让工具内部捕获异常并返回错误说明而不是直接抛出异常。tool def get_current_time() - str: 获取当前系统时间返回格式 yyyy-MM-dd HH:mm:ss。 try: return f{datetime.datetime.now():%Y-%m-%d %H:%M:%S} except Exception as e: return ferror: {e}这样 Observation 是干净的文本模型更容易理解。6.4 验证多轮工具调用实际业务中 Agent 往往需要连续调用多个工具比如先查订单号再查物流状态。可以构造一个“查询商品价格并计算总价”的场景让 Agent 依次调用两个工具。验证重点是第二个工具使用的参数是否来自第一个工具的返回结果这是 Agent 长链路能力的关键指标。7. 接口 API 与批量任务Agent 写好后最常见的落地方式有两种封装成 HTTP API 给前端或其他服务调用批量处理一批问题文本。7.1 用 FastAPI 封装 Agent 服务下面的代码把executor包成一个 POST 接口统一走 JSON 协议。from fastapi import FastAPI from pydantic import BaseModel app FastAPI(titleLangChain Agent API) class ChatRequest(BaseModel): input: str class ChatResponse(BaseModel): output: str app.post(/agent/chat, response_modelChatResponse) async def chat(req: ChatRequest): result executor.invoke({input: req.input}) return ChatResponse(outputresult[output])启动服务uvicorn app:app --host 127.0.0.1 --port 8000然后用 curl 验证接口curl -X POST http://127.0.0.1:8000/agent/chat \ -H Content-Type: application/json \ -d {input: 现在几点了}如果返回内容包含output字段说明接口链路正常。需要特别注意的是executor必须是在app.py模块加载时已经初始化完成不要在请求函数内部重复创建 Agent否则每次请求都会重新加载模型和工具性能和稳定性都会受影响。7.2 Python 批量任务调用批量处理可以简单用循环也可以引入线程池。考虑到 Agent 每个请求都会多次调用大模型先把max_iterations和超时时间限制好再加线程池避免并发过高导致 API 限流。from concurrent.futures import ThreadPoolExecutor questions [ 现在几点了, 11 等于几, 请直接回答LangChain Agent 是什么, ] def safe_invoke(q: str): try: result executor.invoke( {input: q}, config{max_time: 60, callbacks: []}, ) return q, result.get(output) except Exception as e: return q, ffailed: {e} with ThreadPoolExecutor(max_workers2) as pool: results list(pool.map(safe_invoke, questions)) for q, out in results: print(q, -, out)建议生产环境把任务队列、结果持久化和失败重试分开设计不要在一个脚本里堆逻辑。批量任务最好输出成 JSON Lines 文件方便后续审计。{question: 现在几点了, output: 抱歉我暂时无法获取当前时间。}7.3 批量任务配置模板下面是一份项目级配置的 yaml 示例具体字段需要按实际项目调整这里只给通用模板。agent: max_iterations: 5 temperature: 0 verbose: true batch: request_file: ./questions.txt output_file: ./results.jsonl max_workers: 2 timeout: 608. 资源占用与性能观察Agent 本身是一个编排层没有独立的显存开销。资源占用由底层大模型决定接 OpenAI 等云端 API 时本地只做请求转发和文本拼接接 Ollama 本地模型时显存和内存占用取决于模型大小、量化方式和并发数。具体数字需要以本机测试为准。性能上最值得关注的是 token 消耗和轮次。ReAct 循环每一轮都会发起一次大模型调用每轮都会把之前的历史追加进去所以轮数越多token 消耗增长越快。五个轮次和一个轮次的成本差距可能达到数倍。优化顺序应该是先缩减轮次。工具名称和描述尽量准确让模型一次判断正确。再降低单次 token 量。工具返回的 Observation 尽量精简不要返回大段无用日志。最后才考虑并发。并发能提高吞吐但会推高 API 成本或显存占用。如果想观察 token 用量可以通过回调捕获 LLM 的response_metadata。下面是一个简单的回调处理器示例字段名以实际版本为准。from langchain_core.callbacks import BaseCallbackHandler from collections import Counter class TokenHandler(BaseCallbackHandler): def __init__(self): self.usage Counter() def on_llm_end(self, response, **kwargs): message response.generations[0][0].message meta getattr(message, response_metadata, {}) token_usage meta.get(token_usage) if token_usage: self.usage[prompt_tokens] token_usage.get(prompt_tokens, 0) self.usage[completion_tokens] token_usage.get(completion_tokens, 0) handler TokenHandler() result executor.invoke( {input: 现在几点了}, config{callbacks: [handler]}, ) print(token usage:, dict(handler.usage))max_iterations是另一个关键参数。调大可以给模型更多尝试机会但也会放大成本和耗时调小可以快速终止但可能模型还没找到正确答案就被迫结束。建议从 3 开始调观察日志里是否频繁触顶再决定往哪个方向调整。9. 常见问题与排查方法问题现象可能原因排查方式解决方案导入AgentExecutor报错langchain 版本过旧或依赖缺失pip show langchain查看版本升级到 0.2 以上重新安装 langchain-core工具执行后 Agent 直接异常退出工具抛出的异常未被转换为 Observation查看完整堆栈日志在工具内部捕获异常返回错误文本模型反复调用同一个工具工具描述不清晰或 Observation 未帮模型修正判断打开 verbose 日志看每轮 Observation优化工具描述让工具失败信息更明确达到max_iterations仍无结果任务太复杂或超参数太低检查日志是否一直重复 Action提高 max_iterations或拆分提问API 调用超时网络慢或模型推理时间长设置合理的 timeout增大超时时间或换更小的模型本地模型回答质量差模型参数太小查看模型回答换 7B 以上模型或改用云端 APIbase_url无法连接服务商地址配错或网络不可达用 curl 简单验证地址确认 base_url 对应服务商文档批量任务部分失败某个问题触发工具异常或超时查看失败样本的 intermediate_steps增加失败重试对单个问题设置超时排查时优先打开verboseTrue日志会把每一轮的 Thought、Action、Observation 打出来。很多问题不需要调代码看日志就能定位。10. 最佳实践与使用建议工具定义上要做到名称直观、描述准确。描述里应该写清楚“这个工具干什么、什么时候用、参数是什么格式”。模型是根据描述来做推理的描述不准确工具再强大也没用。工具实现上一定要做异常捕获。不要把系统异常直接抛给 Agent最好把错误转换成一句话返回给模型比如“error: 时间服务不可用请重试”。这样 Observation 可读模型才能继续规划下一步。循环控制上必须设置max_iterations和超时时间。没有上限的 ReAct 循环可能造成不可控的成本消耗。生产环境建议把max_iterations控制在 5 以内并且记录每一次 intermediate_steps方便事后回溯。接口服务上Agent 实例要做到单例复用不要在请求内重复创建。对外暴露 API 时建议增加简单的鉴权或 IP 白名单避免服务被随意调用。涉及用户数据和第三方系统时先确认授权范围再做数据脱敏。模型选择上如果业务对响应速度要求高优先使用云端商业模型如果对数据隐私要求高再考虑本地模型。本地模型也要注意显存占用的实际测试不要只看模型参数大小就决定上线。11. 总结与下一步LangChain Agent 的底层并不复杂模型推理、调用工具、拿到反馈、继续推理直到输出最终答案。本文用“不可用工具”这个反例把AgentExecutor的循环过程暴露出来重点是想说明 Observation 对模型推理的影响。工具失败不会让 Agent 崩溃但工具描述错误会让 Agent 走向错误方向。建议你先用 5.1 的异常工具代码跑一遍再把return_intermediate_stepsTrue打开观察每一轮的变化。确认你理解了 ReAct 循环之后再把它换成正常工具对比两种场景下 Agent 的效率和输出质量。下一步扩展方向有三条一是用 LangGraph 替代AgentExecutor通过图结构更精细地控制循环和条件分支二是研究 MCP 工具接入让 Agent 直接使用外部服务能力三是把工具返回值改成 JSON 结构化格式提高模型对复杂结果的利用率。先把这个 ReAct 循环吃透后面做 Agent 开发会顺手很多。