尧图网站建设 
这次我们来看 LangChain Agent 的底层循环。直接说结论Agent 的本质就是一个“推理 → 行动 → 反馈”的循环模型在每一轮里判断要不要调用工具、调用哪个工具、看到结果后再决定下一步。这个概念很多人听过但真正把代码跑起来、亲眼看到循环是怎么走完的是另一回事。这篇文章不打算再讲一遍 ReAct 论文的理论。我会用一个“故意写坏的自定义工具”来演示这个循环。为什么故意写坏因为工具不可用时Agent 不会立刻退出它会拿着错误反馈继续推理、再换动作、再拿反馈循环的痕迹反而暴露得最清楚。看完你就能明白Agent 什么时候调用工具、收到错误后怎么决策、循环在什么条件下结束、token 消耗为什么会在失败场景下快速膨胀。本文的实操内容包括环境准备、用tool自定义工具、构建 ReAct Agent、观察推理-行动-反馈循环的完整日志、把 Agent 封装成 API 接口、批量提问脚本、资源消耗与控制手段、常见问题和排查方式。适合刚开始学 langchain agent 的开发者、想搞清楚 agent 框架底层原理的人、以及准备把 Agent 接到业务系统里的读者。1. 核心能力速览先给一张速览表把这次演示的内容和技术栈固定在同一个坐标系里。能力项说明项目主题LangChain Agent 底层机制推理-行动-反馈循环核心技术范式ReActReasoning Acting即推理与行动交替演示方式自定义一个不可用的工具观察 Agent 的完整循环过程依赖库langchain、langchain-core、langchain-openai模型使用 OpenAI 兼容 API模型需求有 OpenAI 兼容接口即可也可接入本地 Ollama 模型硬件门槛纯 API 调用模式不依赖 GPU本地部署模型时才需要关注显存启动方式Python 脚本运行 / FastAPI 接口服务是否支持 API支持可封装成/agent/run接口是否支持批量任务支持按问题列表循环调用并记录日志核心输出AgentExecutor 的 verbose 日志、中间步骤列表、最终回答适合场景Agent 教学、对话式工具调用、接口封装、失败重试策略设计需要说明的是Agent 本身不承担 LLM 推理它只是一个“编排层”。只要模型服务地址能连通本文的演示在本机跑代码即可完成不涉及显卡、显存、模型文件下载。2. Agent 底层机制推理-行动-反馈循环LangChain Agent 的底层运行逻辑来自于 ReAct 范式。模型的输出被严格约束成一种固定格式包含四种基本内容Thought推理模型解释当前问题说明自己需要什么信息、打算怎么做。Action行动模型选择一个工具名告诉系统“我要调用get_weather”。Action Input行动输入模型给出这个工具的参数例如{city: 北京}。Observation反馈系统执行工具后把返回结果交回给模型作为这一轮的观察结果。整个循环如下用户输入问题。模型输出 Thought Action Action Input。系统调用对应的工具拿到结果。系统把 Thought、Action、Action Input、Observation 追加到上下文里再次发给模型。模型基于新的观察结果继续输出 Thought Action或者直接输出 Final Answer。循环直到模型输出 Final Answer或触发人为设置的上限。关键点在于模型在每一步之间并不是“有状态”的。它每次都只看到一份不断变长的对话历史没有任何内部记忆。所谓“Agent 记得之前失败过”只是因为失败记录被拼进了下一次输入的上下文里。这也是为什么循环看起来是重复的、可追踪的、消耗 token 的。循环的终止条件有三个来源模型自己输出Final Answer正常结束。达到max_iterations执行器强制停止。超过max_execution_time执行器按超时处理。如果工具输出解析失败而配置了handle_parsing_errorsTrue解析错误本身也会变成一条 Observation 送回给模型相当于又触发一轮循环。这个细节在后面的排查章节还会用到。补充一点新版 LangChain 已经把 Agent 的运行时逐渐迁移到 LangGraph 上官方推荐用create_agent或 LangGraph 的 StateGraph 写更可控的循环。但AgentExecutor的 verbose 日志格式仍然是最直观的教学工具。先把这里的循环看明白再去看 LangGraph 的状态图会顺畅很多。3. 环境准备与依赖安装本文演示基于 Python建议使用 Python 3.10 或更高版本。先创建虚拟环境并安装依赖。python -m venv .venv source .venv/bin/activate # Windows 下执行 .venv\Scripts\activate pip install -U langchain langchain-core langchain-openai pip install python-dotenv如果后续要跑接口服务再安装 FastAPI 和 uvicorn。pip install fastapi uvicorn模型端我建议先使用 OpenAI 兼容 API。很多模型厂商都提供这个协议比如 DeepSeek、通义千问、Kimi 等。这样本地不需要下载模型文件也不需要 GPU。配置方式很简单用.env文件统一管理密钥和接口地址。OPENAI_API_KEYsk-你的密钥 OPENAI_BASE_URLhttps://api.deepseek.com/v1 LLM_MODELdeepseek-chat如果你用的是其他厂商把base_url和model换成对应平台的地址和模型名即可。如果是本地 Ollamabase_url可以填http://localhost:11434/v1model填你本地拉取的模型名。需要注意的是一旦换成本地小模型Agent 的输出格式稳定性可能下降解析失败的概率会明显高于大模型。4. 自定义工具故意写一个不可用的工具在 LangChain 里自定义工具最推荐的方式是使用tool装饰器。工具名默认是函数名工具描述默认是函数的 docstring。这里有个新手最容易踩的坑工具描述就是“模型能看到的功能说明书”。如果 docstring 为空或者写得含糊模型很可能根本不会调用这个工具。先看一个正常的工具用于获取当前日期。import datetime from langchain_core.tools import tool tool def get_today_date() - str: 返回今天的日期格式为 YYYY-MM-DD。 return datetime.date.today().isoformat()再看本文的主角一个“故意写坏”的天气工具。它不接任何真实天气服务而是直接返回一条错误字符串。这样做的效果是Agent 每次调用它都会拿到一条不可用的 Observation从而被迫继续推理。tool def get_weather(city: str) - str: 根据城市名查询实时天气返回天气现象和温度。 # 演示用故意不接真实天气服务直接返回错误。 # 这一条错误就是 Agent 拿到的 Observation。 return fERROR: weather service timeout, city{city}再提供一个抛异常的版本。当工具抛出ToolException时如果执行器配置了handle_tool_errorTrue异常信息会被包装成 Observation 送回给模型而不是直接让程序崩溃。from langchain_core.exceptions import ToolException tool def query_stock(code: str) - str: 查询 A 股股票最新价格。 raise ToolException(f行情接口 500无法获取 {code} 的价格)构造工具列表时把正常工具和坏工具放在一起tools [get_today_date, get_weather, query_stock]这里要理解“返回错误字符串”和“抛异常”两种写法的区别。返回错误字符串更接近真实业务很多 HTTP 接口超时、限流、参数错误时调用方拿到的是业务错误码而不是异常。抛异常更接近程序层面的故障网络断开、JSON 解析失败。两种情况下Agent 的反馈内容不一样后续的决策路径也会不一样。这正是观察循环行为最好的素材。5. 构建 Agent 并启动演示有了工具接下来构建 ReAct Agent。先写好 ReAct 提示词模板模板里必须包含{tools}、{tool_names}、{input}、{agent_scratchpad}这四个占位符create_react_agent会负责填充它们。import os from dotenv import load_dotenv from langchain.agents import AgentExecutor, create_react_agent from langchain_core.prompts import PromptTemplate from langchain_openai import ChatOpenAI load_dotenv() llm ChatOpenAI( modelos.getenv(LLM_MODEL, deepseek-chat), api_keyos.getenv(OPENAI_API_KEY), base_urlos.getenv(OPENAI_BASE_URL, https://api.deepseek.com/v1), temperature0, ) react_prompt PromptTemplate.from_template( Answer the following questions as best you can. You have access to the following tools: {tools} Use the following format: 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} ) agent create_react_agent(llmllm, toolstools, promptreact_prompt) executor AgentExecutor( agentagent, toolstools, verboseTrue, return_intermediate_stepsTrue, max_iterations6, max_execution_time60, handle_parsing_errorsTrue, handle_tool_errorTrue, ) result executor.invoke({input: 今天是什么日期顺便帮我查一下北京的天气。}) print(最终回答, result[output]) print(中间步骤) for i, (action, observation) in enumerate(result.get(intermediate_steps, []), 1): print(f[{i}] 工具{action.tool}参数{action.tool_input}) print(f 观察{observation})这里有几个参数需要解释verboseTrue在终端打印完整的 Thought / Action / Action Input / Observation 日志这是观察循环的核心手段。return_intermediate_stepsTrue把中间步骤以结构化列表返回方便后续在代码里统计循环轮数。max_iterations6限制最大循环轮数防止模型反复重试导致无限循环。handle_parsing_errorsTrue模型输出格式不正确时把解析错误当作反馈送回模型。handle_tool_errorTrue工具抛出ToolException时把异常信息当作反馈送回模型。6. 观察循环一次失败演示的完整日志运行上面的代码verbose 日志大致如下。不同模型的措辞会有差异但结构一定是这样的循环 Entering new AgentExecutor chain... Thought: 用户问了两件事今天的日期和北京的天气。日期可以直接用 get_today_date 工具获得。 Action: get_today_date Action Input: {} Observation: 2025-06-10 Thought: 日期已经拿到。接下来需要查询北京的天气。 Action: get_weather Action Input: {city: 北京} Observation: ERROR: weather service timeout, city北京 Thought: 天气工具返回了超时错误可能只是暂时的问题我重试一次。 Action: get_weather Action Input: {city: 北京} Observation: ERROR: weather service timeout, city北京 Thought: 再次超时说明天气服务当前不可用。我无法获取真实天气应如实告知用户。 Final Answer: 今天是 2025-06-10。北京的天气暂时无法查询因为天气服务超时请稍后再试。 Finished chain.这段日志是整个文章最有价值的部分。我们现在把它拆开看。第一轮循环模型先输出 Thought说明它识别出需要日期信息然后 Action 选了get_today_dateAction Input 是空对象。工具正常返回日期Observation 是2025-06-10。这是正常工具的工作链路只有一轮。第二轮循环模型拿到日期后Thought 说“接下来需要查询北京的天气”Action 选了get_weather参数是{city: 北京}。可是工具坏了Observation 是超时错误。注意模型并没有直接放弃而是把“超时”理解成“可能是暂时的问题”。这是模型在工具不可用时最常见的第一反应重试。第三轮循环模型再次调用同一个工具再次拿到同样的超时错误。这时候模型的判断发生了变化它意识到“再次超时说明服务不可用”于是不再盲目重试而是选择输出 Final Answer并如实告诉用户天气暂时查不了。这个例子说明三件事第一循环是“观察驱动”的。模型每一步的输出都依赖上一步的 Observation。工具返回的错误字符串本身也是观察结果而且它的措辞会影响模型下一步的选择。如果让get_weather返回的不是timeout而是city not found模型大概率不会重试而是会换个城市名或直接结束。第二模型的重试策略不是写死的。它没有固定的“失败重试三次”逻辑而是根据错误信息现场推理。这就是 Agent 和普通函数式调用流程最本质的区别。第三循环是有代价的。一次失败重试就意味着多一次完整的 LLM 调用多送一截上下文给模型。坏工具会把 token 消耗成倍放大。这个问题在第 8 节还会展开。如果把工具换成query_stock也就是抛异常的版本执行器会捕获ToolException并把异常信息包装成 Observation。日志里的 Observation 会变成类似Error: 行情接口 500无法获取 600519 的价格。模型同样会基于这条反馈继续推理。这就是handle_tool_errorTrue的作用把程序异常也变成 Agent 可理解的观察结果而不是中断整个链。需要提醒的是模型是否重试、重试几次、什么时候放弃和模型本身的能力高度相关。大模型通常能正确识别“第二次还是同样的超时应该停止”但本地小模型可能会反复重试同一参数直到撞上max_iterations才停下来。所以测试时一定要设置上限。7. 把 Agent 封装成接口服务Agent 的循环过程适合在脚本里观察但落到业务系统里通常要把它封装成 HTTP 接口。这里给一个 FastAPI 封装示例。# agent_server.py from fastapi import FastAPI from pydantic import BaseModel from langchain.agents import AgentExecutor, create_react_agent from langchain_core.prompts import PromptTemplate from langchain_openai import ChatOpenAI from langchain_core.tools import tool import os app FastAPI(titleLangChain Agent Demo) llm ChatOpenAI( modelos.getenv(LLM_MODEL, deepseek-chat), api_keyos.getenv(OPENAI_API_KEY), base_urlos.getenv(OPENAI_BASE_URL, https://api.deepseek.com/v1), temperature0, ) tool def get_weather(city: str) - str: 根据城市名查询实时天气返回天气现象和温度。 return fERROR: weather service timeout, city{city} AGENT_TOOLS [get_weather] react_prompt PromptTemplate.from_template( Answer the following questions as best you can. You have access to the following tools: {tools} Use the following format: 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} ) def run_agent_once(question: str, max_iterations: int 6) - dict: agent create_react_agent(llmllm, toolsAGENT_TOOLS, promptreact_prompt) executor AgentExecutor( agentagent, toolsAGENT_TOOLS, max_iterationsmax_iterations, max_execution_time60, handle_parsing_errorsTrue, handle_tool_errorTrue, ) return executor.invoke({input: question}) class AgentRequest(BaseModel): question: str max_iterations: int 6 class AgentStep(BaseModel): tool: str tool_input: object observation: str class AgentResponse(BaseModel): answer: str steps: list[AgentStep] app.post(/agent/run, response_modelAgentResponse) def agent_run(req: AgentRequest): result run_agent_once(req.question, req.max_iterations) steps [] for action, observation in result.get(intermediate_steps, []): steps.append(AgentStep( toolaction.tool, tool_inputaction.tool_input, observationstr(observation), )) return AgentResponse(answerresult[output], stepssteps)启动服务uvicorn agent_server:app --host 127.0.0.1 --port 8000用 curl 验证curl -X POST http://127.0.0.1:8000/agent/run \ -H Content-Type: application/json \ -d {question: 查一下北京的天气} \ --max-time 90返回结构里既有最终答案也有中间步骤。中间步骤就是 Agent 的循环记录业务系统可以用它做审计和日志展示。这里有一个工程细节必须提醒Agent 单次请求内部会有多次 LLM 调用耗时通常是几十秒级别远高于普通 HTTP 接口。FastAPI 的同步def接口运行在线程池里演示没问题但如果要高并发需要把 Agent 调用改成异步执行或者把请求丢进任务队列前端轮询结果。在接口层面一定要设置调用超时不要把 Agent 的耗时无限放大到网关层。批量提问的思路类似维护一个问题列表逐个调用run_agent_once每个问题设置独立的最大迭代数和超时失败时记录日志继续下一个而不是让整个批次卡死。questions [ 今天日期是什么, 北京的天气怎么样, 查一下 600519 的股价, ] for q in questions: try: res run_agent_once(q, max_iterations4) print(Q:, q) print(A:, res[output]) print( 循环轮数:, len(res.get(intermediate_steps, []))) except Exception as e: print(Q:, q, 失败:, e)8. 资源消耗与性能观察失败循环的电费账单Agent 的资源消耗瓶颈不在 CPU 和显卡而在 token。每一轮循环LLM 都会收到一份新增的上下文Thought、Action、Action Input、Observation再加上工具描述、系统提示词、历史轮次全部都要重新计算。循环 N 次上下文长度近似线性增长成本也近似线性放大。坏工具会直接放大这个成本。因为模型在收到“超时”这类可恢复错误时第一反应往往是重试。一次重试 多一次 LLM 调用。如果模型连续重试三次才放弃那就等于这次提问烧掉了四次请求的 token而用户只得到一句“天气暂时查不了”。观察 token 消耗的办法很朴素在run_agent_once里打印模型调用信息。如果使用langchain_openai可以在ChatOpenAI上开启callback或者直接使用 LangSmith 追踪。不想引入额外依赖时最简单的方式就是数循环轮数len(result.get(intermediate_steps, []))就是循环次数每轮至少包含一次模型调用。轮数越多token 消耗越大。控制消耗的方法有几种。第一设置max_iterations。这是最直接的刹车。根据业务场景决定一次提问最多允许模型调用几次工具不要使用无限迭代。第二限制工具数量。工具越多模型选错工具的概率越高交叉重试的可能性越大。先把工具数量控制在 5 个以内跑通核心流程再逐步加。第三收敛工具描述。工具描述越精炼每轮输入的固定 token 越少。没有人会在描述里写三段论文模型只需要知道“这个工具干什么、参数是什么、什么情况会返回错误”。第四设计“可区分”的错误消息。如果工具的失败信息总是同一句话模型就无法判断“这次失败和上次失败是不是同一个原因”。把错误结构化成{status: error, reason: timeout, retryable: true}模型就能更准确地决定是否重试。第五本地模型部署时才需要关注显存和推理速度。对于 API LangChain Agent底层循环解析：用坏工具看清推理-行动-反馈机制
这次我们来看 LangChain Agent 的底层循环。直接说结论Agent 的本质就是一个“推理 → 行动 → 反馈”的循环模型在每一轮里判断要不要调用工具、调用哪个工具、看到结果后再决定下一步。这个概念很多人听过但真正把代码跑起来、亲眼看到循环是怎么走完的是另一回事。这篇文章不打算再讲一遍 ReAct 论文的理论。我会用一个“故意写坏的自定义工具”来演示这个循环。为什么故意写坏因为工具不可用时Agent 不会立刻退出它会拿着错误反馈继续推理、再换动作、再拿反馈循环的痕迹反而暴露得最清楚。看完你就能明白Agent 什么时候调用工具、收到错误后怎么决策、循环在什么条件下结束、token 消耗为什么会在失败场景下快速膨胀。本文的实操内容包括环境准备、用tool自定义工具、构建 ReAct Agent、观察推理-行动-反馈循环的完整日志、把 Agent 封装成 API 接口、批量提问脚本、资源消耗与控制手段、常见问题和排查方式。适合刚开始学 langchain agent 的开发者、想搞清楚 agent 框架底层原理的人、以及准备把 Agent 接到业务系统里的读者。1. 核心能力速览先给一张速览表把这次演示的内容和技术栈固定在同一个坐标系里。能力项说明项目主题LangChain Agent 底层机制推理-行动-反馈循环核心技术范式ReActReasoning Acting即推理与行动交替演示方式自定义一个不可用的工具观察 Agent 的完整循环过程依赖库langchain、langchain-core、langchain-openai模型使用 OpenAI 兼容 API模型需求有 OpenAI 兼容接口即可也可接入本地 Ollama 模型硬件门槛纯 API 调用模式不依赖 GPU本地部署模型时才需要关注显存启动方式Python 脚本运行 / FastAPI 接口服务是否支持 API支持可封装成/agent/run接口是否支持批量任务支持按问题列表循环调用并记录日志核心输出AgentExecutor 的 verbose 日志、中间步骤列表、最终回答适合场景Agent 教学、对话式工具调用、接口封装、失败重试策略设计需要说明的是Agent 本身不承担 LLM 推理它只是一个“编排层”。只要模型服务地址能连通本文的演示在本机跑代码即可完成不涉及显卡、显存、模型文件下载。2. Agent 底层机制推理-行动-反馈循环LangChain Agent 的底层运行逻辑来自于 ReAct 范式。模型的输出被严格约束成一种固定格式包含四种基本内容Thought推理模型解释当前问题说明自己需要什么信息、打算怎么做。Action行动模型选择一个工具名告诉系统“我要调用get_weather”。Action Input行动输入模型给出这个工具的参数例如{city: 北京}。Observation反馈系统执行工具后把返回结果交回给模型作为这一轮的观察结果。整个循环如下用户输入问题。模型输出 Thought Action Action Input。系统调用对应的工具拿到结果。系统把 Thought、Action、Action Input、Observation 追加到上下文里再次发给模型。模型基于新的观察结果继续输出 Thought Action或者直接输出 Final Answer。循环直到模型输出 Final Answer或触发人为设置的上限。关键点在于模型在每一步之间并不是“有状态”的。它每次都只看到一份不断变长的对话历史没有任何内部记忆。所谓“Agent 记得之前失败过”只是因为失败记录被拼进了下一次输入的上下文里。这也是为什么循环看起来是重复的、可追踪的、消耗 token 的。循环的终止条件有三个来源模型自己输出Final Answer正常结束。达到max_iterations执行器强制停止。超过max_execution_time执行器按超时处理。如果工具输出解析失败而配置了handle_parsing_errorsTrue解析错误本身也会变成一条 Observation 送回给模型相当于又触发一轮循环。这个细节在后面的排查章节还会用到。补充一点新版 LangChain 已经把 Agent 的运行时逐渐迁移到 LangGraph 上官方推荐用create_agent或 LangGraph 的 StateGraph 写更可控的循环。但AgentExecutor的 verbose 日志格式仍然是最直观的教学工具。先把这里的循环看明白再去看 LangGraph 的状态图会顺畅很多。3. 环境准备与依赖安装本文演示基于 Python建议使用 Python 3.10 或更高版本。先创建虚拟环境并安装依赖。python -m venv .venv source .venv/bin/activate # Windows 下执行 .venv\Scripts\activate pip install -U langchain langchain-core langchain-openai pip install python-dotenv如果后续要跑接口服务再安装 FastAPI 和 uvicorn。pip install fastapi uvicorn模型端我建议先使用 OpenAI 兼容 API。很多模型厂商都提供这个协议比如 DeepSeek、通义千问、Kimi 等。这样本地不需要下载模型文件也不需要 GPU。配置方式很简单用.env文件统一管理密钥和接口地址。OPENAI_API_KEYsk-你的密钥 OPENAI_BASE_URLhttps://api.deepseek.com/v1 LLM_MODELdeepseek-chat如果你用的是其他厂商把base_url和model换成对应平台的地址和模型名即可。如果是本地 Ollamabase_url可以填http://localhost:11434/v1model填你本地拉取的模型名。需要注意的是一旦换成本地小模型Agent 的输出格式稳定性可能下降解析失败的概率会明显高于大模型。4. 自定义工具故意写一个不可用的工具在 LangChain 里自定义工具最推荐的方式是使用tool装饰器。工具名默认是函数名工具描述默认是函数的 docstring。这里有个新手最容易踩的坑工具描述就是“模型能看到的功能说明书”。如果 docstring 为空或者写得含糊模型很可能根本不会调用这个工具。先看一个正常的工具用于获取当前日期。import datetime from langchain_core.tools import tool tool def get_today_date() - str: 返回今天的日期格式为 YYYY-MM-DD。 return datetime.date.today().isoformat()再看本文的主角一个“故意写坏”的天气工具。它不接任何真实天气服务而是直接返回一条错误字符串。这样做的效果是Agent 每次调用它都会拿到一条不可用的 Observation从而被迫继续推理。tool def get_weather(city: str) - str: 根据城市名查询实时天气返回天气现象和温度。 # 演示用故意不接真实天气服务直接返回错误。 # 这一条错误就是 Agent 拿到的 Observation。 return fERROR: weather service timeout, city{city}再提供一个抛异常的版本。当工具抛出ToolException时如果执行器配置了handle_tool_errorTrue异常信息会被包装成 Observation 送回给模型而不是直接让程序崩溃。from langchain_core.exceptions import ToolException tool def query_stock(code: str) - str: 查询 A 股股票最新价格。 raise ToolException(f行情接口 500无法获取 {code} 的价格)构造工具列表时把正常工具和坏工具放在一起tools [get_today_date, get_weather, query_stock]这里要理解“返回错误字符串”和“抛异常”两种写法的区别。返回错误字符串更接近真实业务很多 HTTP 接口超时、限流、参数错误时调用方拿到的是业务错误码而不是异常。抛异常更接近程序层面的故障网络断开、JSON 解析失败。两种情况下Agent 的反馈内容不一样后续的决策路径也会不一样。这正是观察循环行为最好的素材。5. 构建 Agent 并启动演示有了工具接下来构建 ReAct Agent。先写好 ReAct 提示词模板模板里必须包含{tools}、{tool_names}、{input}、{agent_scratchpad}这四个占位符create_react_agent会负责填充它们。import os from dotenv import load_dotenv from langchain.agents import AgentExecutor, create_react_agent from langchain_core.prompts import PromptTemplate from langchain_openai import ChatOpenAI load_dotenv() llm ChatOpenAI( modelos.getenv(LLM_MODEL, deepseek-chat), api_keyos.getenv(OPENAI_API_KEY), base_urlos.getenv(OPENAI_BASE_URL, https://api.deepseek.com/v1), temperature0, ) react_prompt PromptTemplate.from_template( Answer the following questions as best you can. You have access to the following tools: {tools} Use the following format: 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} ) agent create_react_agent(llmllm, toolstools, promptreact_prompt) executor AgentExecutor( agentagent, toolstools, verboseTrue, return_intermediate_stepsTrue, max_iterations6, max_execution_time60, handle_parsing_errorsTrue, handle_tool_errorTrue, ) result executor.invoke({input: 今天是什么日期顺便帮我查一下北京的天气。}) print(最终回答, result[output]) print(中间步骤) for i, (action, observation) in enumerate(result.get(intermediate_steps, []), 1): print(f[{i}] 工具{action.tool}参数{action.tool_input}) print(f 观察{observation})这里有几个参数需要解释verboseTrue在终端打印完整的 Thought / Action / Action Input / Observation 日志这是观察循环的核心手段。return_intermediate_stepsTrue把中间步骤以结构化列表返回方便后续在代码里统计循环轮数。max_iterations6限制最大循环轮数防止模型反复重试导致无限循环。handle_parsing_errorsTrue模型输出格式不正确时把解析错误当作反馈送回模型。handle_tool_errorTrue工具抛出ToolException时把异常信息当作反馈送回模型。6. 观察循环一次失败演示的完整日志运行上面的代码verbose 日志大致如下。不同模型的措辞会有差异但结构一定是这样的循环 Entering new AgentExecutor chain... Thought: 用户问了两件事今天的日期和北京的天气。日期可以直接用 get_today_date 工具获得。 Action: get_today_date Action Input: {} Observation: 2025-06-10 Thought: 日期已经拿到。接下来需要查询北京的天气。 Action: get_weather Action Input: {city: 北京} Observation: ERROR: weather service timeout, city北京 Thought: 天气工具返回了超时错误可能只是暂时的问题我重试一次。 Action: get_weather Action Input: {city: 北京} Observation: ERROR: weather service timeout, city北京 Thought: 再次超时说明天气服务当前不可用。我无法获取真实天气应如实告知用户。 Final Answer: 今天是 2025-06-10。北京的天气暂时无法查询因为天气服务超时请稍后再试。 Finished chain.这段日志是整个文章最有价值的部分。我们现在把它拆开看。第一轮循环模型先输出 Thought说明它识别出需要日期信息然后 Action 选了get_today_dateAction Input 是空对象。工具正常返回日期Observation 是2025-06-10。这是正常工具的工作链路只有一轮。第二轮循环模型拿到日期后Thought 说“接下来需要查询北京的天气”Action 选了get_weather参数是{city: 北京}。可是工具坏了Observation 是超时错误。注意模型并没有直接放弃而是把“超时”理解成“可能是暂时的问题”。这是模型在工具不可用时最常见的第一反应重试。第三轮循环模型再次调用同一个工具再次拿到同样的超时错误。这时候模型的判断发生了变化它意识到“再次超时说明服务不可用”于是不再盲目重试而是选择输出 Final Answer并如实告诉用户天气暂时查不了。这个例子说明三件事第一循环是“观察驱动”的。模型每一步的输出都依赖上一步的 Observation。工具返回的错误字符串本身也是观察结果而且它的措辞会影响模型下一步的选择。如果让get_weather返回的不是timeout而是city not found模型大概率不会重试而是会换个城市名或直接结束。第二模型的重试策略不是写死的。它没有固定的“失败重试三次”逻辑而是根据错误信息现场推理。这就是 Agent 和普通函数式调用流程最本质的区别。第三循环是有代价的。一次失败重试就意味着多一次完整的 LLM 调用多送一截上下文给模型。坏工具会把 token 消耗成倍放大。这个问题在第 8 节还会展开。如果把工具换成query_stock也就是抛异常的版本执行器会捕获ToolException并把异常信息包装成 Observation。日志里的 Observation 会变成类似Error: 行情接口 500无法获取 600519 的价格。模型同样会基于这条反馈继续推理。这就是handle_tool_errorTrue的作用把程序异常也变成 Agent 可理解的观察结果而不是中断整个链。需要提醒的是模型是否重试、重试几次、什么时候放弃和模型本身的能力高度相关。大模型通常能正确识别“第二次还是同样的超时应该停止”但本地小模型可能会反复重试同一参数直到撞上max_iterations才停下来。所以测试时一定要设置上限。7. 把 Agent 封装成接口服务Agent 的循环过程适合在脚本里观察但落到业务系统里通常要把它封装成 HTTP 接口。这里给一个 FastAPI 封装示例。# agent_server.py from fastapi import FastAPI from pydantic import BaseModel from langchain.agents import AgentExecutor, create_react_agent from langchain_core.prompts import PromptTemplate from langchain_openai import ChatOpenAI from langchain_core.tools import tool import os app FastAPI(titleLangChain Agent Demo) llm ChatOpenAI( modelos.getenv(LLM_MODEL, deepseek-chat), api_keyos.getenv(OPENAI_API_KEY), base_urlos.getenv(OPENAI_BASE_URL, https://api.deepseek.com/v1), temperature0, ) tool def get_weather(city: str) - str: 根据城市名查询实时天气返回天气现象和温度。 return fERROR: weather service timeout, city{city} AGENT_TOOLS [get_weather] react_prompt PromptTemplate.from_template( Answer the following questions as best you can. You have access to the following tools: {tools} Use the following format: 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} ) def run_agent_once(question: str, max_iterations: int 6) - dict: agent create_react_agent(llmllm, toolsAGENT_TOOLS, promptreact_prompt) executor AgentExecutor( agentagent, toolsAGENT_TOOLS, max_iterationsmax_iterations, max_execution_time60, handle_parsing_errorsTrue, handle_tool_errorTrue, ) return executor.invoke({input: question}) class AgentRequest(BaseModel): question: str max_iterations: int 6 class AgentStep(BaseModel): tool: str tool_input: object observation: str class AgentResponse(BaseModel): answer: str steps: list[AgentStep] app.post(/agent/run, response_modelAgentResponse) def agent_run(req: AgentRequest): result run_agent_once(req.question, req.max_iterations) steps [] for action, observation in result.get(intermediate_steps, []): steps.append(AgentStep( toolaction.tool, tool_inputaction.tool_input, observationstr(observation), )) return AgentResponse(answerresult[output], stepssteps)启动服务uvicorn agent_server:app --host 127.0.0.1 --port 8000用 curl 验证curl -X POST http://127.0.0.1:8000/agent/run \ -H Content-Type: application/json \ -d {question: 查一下北京的天气} \ --max-time 90返回结构里既有最终答案也有中间步骤。中间步骤就是 Agent 的循环记录业务系统可以用它做审计和日志展示。这里有一个工程细节必须提醒Agent 单次请求内部会有多次 LLM 调用耗时通常是几十秒级别远高于普通 HTTP 接口。FastAPI 的同步def接口运行在线程池里演示没问题但如果要高并发需要把 Agent 调用改成异步执行或者把请求丢进任务队列前端轮询结果。在接口层面一定要设置调用超时不要把 Agent 的耗时无限放大到网关层。批量提问的思路类似维护一个问题列表逐个调用run_agent_once每个问题设置独立的最大迭代数和超时失败时记录日志继续下一个而不是让整个批次卡死。questions [ 今天日期是什么, 北京的天气怎么样, 查一下 600519 的股价, ] for q in questions: try: res run_agent_once(q, max_iterations4) print(Q:, q) print(A:, res[output]) print( 循环轮数:, len(res.get(intermediate_steps, []))) except Exception as e: print(Q:, q, 失败:, e)8. 资源消耗与性能观察失败循环的电费账单Agent 的资源消耗瓶颈不在 CPU 和显卡而在 token。每一轮循环LLM 都会收到一份新增的上下文Thought、Action、Action Input、Observation再加上工具描述、系统提示词、历史轮次全部都要重新计算。循环 N 次上下文长度近似线性增长成本也近似线性放大。坏工具会直接放大这个成本。因为模型在收到“超时”这类可恢复错误时第一反应往往是重试。一次重试 多一次 LLM 调用。如果模型连续重试三次才放弃那就等于这次提问烧掉了四次请求的 token而用户只得到一句“天气暂时查不了”。观察 token 消耗的办法很朴素在run_agent_once里打印模型调用信息。如果使用langchain_openai可以在ChatOpenAI上开启callback或者直接使用 LangSmith 追踪。不想引入额外依赖时最简单的方式就是数循环轮数len(result.get(intermediate_steps, []))就是循环次数每轮至少包含一次模型调用。轮数越多token 消耗越大。控制消耗的方法有几种。第一设置max_iterations。这是最直接的刹车。根据业务场景决定一次提问最多允许模型调用几次工具不要使用无限迭代。第二限制工具数量。工具越多模型选错工具的概率越高交叉重试的可能性越大。先把工具数量控制在 5 个以内跑通核心流程再逐步加。第三收敛工具描述。工具描述越精炼每轮输入的固定 token 越少。没有人会在描述里写三段论文模型只需要知道“这个工具干什么、参数是什么、什么情况会返回错误”。第四设计“可区分”的错误消息。如果工具的失败信息总是同一句话模型就无法判断“这次失败和上次失败是不是同一个原因”。把错误结构化成{status: error, reason: timeout, retryable: true}模型就能更准确地决定是否重试。第五本地模型部署时才需要关注显存和推理速度。对于 API