本文聚焦 LangChain Agent 系统的扩展能力：如何自定义工具（Tool）、编写插件，以及如何灵活管理 Agent 的中间推理状态（
AgentAction/AgentFinish）。目标是帮助你搭建高可控、可复用的 Agent 工具体系。
一、自定义 Tool 的标准方式：继承 BaseTool
默认装饰器封装适合简单场景，但复杂业务时推荐继承 BaseTool：
from langchain_core.tools import BaseToolclass ReverseTool(BaseTool): name: str = "reverse" description: str = "反转输入字符串，例如 'hello' → 'olleh'" def _run(self, query: str) -> str: return query[::-1] async def _arun(self, query: str) -> str: raise NotImplementedError("Async not implemented")reverse_tool = ReverseTool()print(reverse_tool.invoke("我是一个正向的字符串。"))# 输出：。串符字的向正个一是我✅ 使用
invoke()即可获取同步结果。
二、AgentPlugin 插件：插入 Agent 执行流程
先创建一个 ReAct Agent
import osfrom langchain_core.prompts import PromptTemplatefrom langchain.agents import create_react_agentfrom langchain_openai import ChatOpenAI# 初始化模型llm = ChatOpenAI( temperature=0, model='deepseek-chat', openai_api_key=os.getenv('DEEPSEEK_API_KEY'), openai_api_base='https://api.deepseek.com', max_tokens=1024)template = """你是一个可以通过工具来完成任务的智能 AI 助手，擅长推理、规划和执行。你的目标是尽可能准确地解决用户的问题。可能的情况下请优先使用以下工具来完成任务：{tools}严格地使用以下格式进行思考与行动：Question: 用户的输入内容Thought: 你对当前问题的理解与推理Action: 选择调用的工具名称，应为 [{tool_names}] 之一Action Input: 务必准确的工具所需的输入Observation: 调用工具返回的结果...（Thought/Action/Action Input/Observation 可以重复多次）Thought: 根据观察得出的最终结论Final Answer: 给用户的直接答复开始！Question: {input}Thought: {agent_scratchpad}"""prompt = PromptTemplate.from_template(template)# 构造 ReAct 类型的 Agentagent = create_react_agent(llm=llm, tools=[ReverseTool()], prompt=prompt)LangChain 支持插件机制：插件可以拦截、修改行为或日志。例如插入日志追踪插件：
from langchain.agents import AgentExecutorfrom langchain.callbacks.base import BaseCallbackHandlerclass LoggingPlugin(BaseCallbackHandler): def on_agent_action(self, action, **kwargs): print(f"Agent 决策：调用工具 {action.tool}，输入 {action.tool_input}") def on_tool_end(self, output, **kwargs): print("工具返回：", output)agent_executor = AgentExecutor.from_agent_and_tools( agent=agent, tools=[ReverseTool(callbacks=[LoggingPlugin()])], callbacks=[LoggingPlugin()], verbose=False, handle_parsing_errors=True)agent_executor.invoke({"input": "请你优先使用我提供的工具反转字符串：LangChain"})插件会在每轮 AgentAction 与工具执行后打印日志。
三、AgentAction 与 AgentFinish：控制 Agent 行为流程
在 LangChain 的 Agent 执行过程中，核心控制流依赖两个数据结构：
AgentAction：表示 下一步调用哪个工具、用什么参数、以及对应的思考日志。AgentFinish：表示 Agent 推理结束，输出最终结果。🔹1. AgentAction：代理调用工具的一步行为
每一轮推理中，Agent 可能会产生一个 AgentAction，结构如下：
AgentAction( tool="reverse", # 工具名称（必须与 tools 列表中的名称匹配） tool_input="LangChain", # 传给工具的输入 log="我认为需要调用 reverse 工具将字符串反转。" # Scratchpad 中的中间推理日志)✅ 注意：这个
log不只是为了输出，它将被用于下一轮 prompt 的 scratchpad 填充，指导 LLM 的进一步思考。
🔹2. AgentFinish：代理任务完成的信号
当 Agent 认为任务完成时，会返回一个 AgentFinish 对象：
AgentFinish( return_values={"output": "niahCgnaL"}, # 最终输出（dict） log="工具调用完毕，任务完成。" # 可用于展示最终思考过程)AgentExecutor 一旦收到 AgentFinish，就会 终止循环并返回结果。
🔹3. 在 Agent 控制循环中的作用
LangChain 中 AgentExecutor 的主循环本质上是这样的伪代码：
while True: action_or_finish = agent.plan(...) if isinstance(action_or_finish, AgentFinish): return action_or_finish.return_values elif isinstance(action_or_finish, AgentAction): observation = tool.run(action_or_finish.tool_input) intermediate_steps.append((action_or_finish, observation))其中 intermediate_steps 会自动构造 scratchpad，供下一轮推理参考。
源码：
langchain.agents.agent.AgentExecutor._call()
🔹4. 你可以自定义这两个输出：用于测试 / 调试 / 插件式控制
# 手动构造一个完整的“伪 Agent”行为流程steps = [ AgentAction(tool="reverse", tool_input="LangChain", log="尝试反转字符串"), AgentFinish(return_values={"output": "niahCgnaL"}, log="得到最终反转结果")]for step in steps: if isinstance(step, AgentAction): print(f"调用工具：{step.tool}，输入：{step.tool_input}") elif isinstance(step, AgentFinish): print(f"完成推理，输出：{step.return_values['output']}")也可以在自定义 Agent 类中显式控制输出：
from langchain.agents import BaseSingleActionAgentclass MyManualAgent(BaseSingleActionAgent): def plan(self, intermediate_steps, **kwargs): return AgentAction( tool="my_tool", tool_input="my_input", log="我决定调用 my_tool" )🔹5. AgentAction 与 Scratchpad 的关系
LangChain 会将 intermediate_steps: List[Tuple[AgentAction, str]] 变为 scratchpad：
Thought: 我需要调用 reverse 工具来处理输入Action: reverseAction Input: LangChainObservation: niahCgnaL这个会拼接到下一轮 prompt 中，引导大模型思考。
🔹6. 实战用途场景
| 场景 | 使用方式 |
|---|---|
| ✅ 单元测试 | 手动构造 AgentAction / AgentFinish 检查工具行为与响应 |
| ✅ 多 Agent 协同 | 每个 Agent 按返回 AgentAction 协调工具调用 |
| ✅ 控制工具调用 | 你可以中途插入自定义 AgentAction 实现特定路径控制 |
| ✅ 可解释性分析 | AgentAction.log 和 AgentFinish.log 记录了完整推理过程 |
| ✅ 自定义插件 | 插件可基于这两个数据结构实现工具路由、后处理等能力 |
import osfrom typing import List, Tuple, Any, Sequencefrom langchain.agents import BaseSingleActionAgent, AgentExecutorfrom langchain_core.agents import AgentAction, AgentFinishfrom langchain.schema import SystemMessage, HumanMessagefrom langchain.tools import Toolfrom langchain_openai import ChatOpenAI# 1️⃣ 工具定义def reverse_string(s: str) -> str: return s[::-1]tools = [ Tool( name="reverse", func=reverse_string, description="反转字符串" )]# 2️⃣ LLM 实例llm = ChatOpenAI( temperature=0.1, model='deepseek-chat', openai_api_key=os.getenv('DEEPSEEK_API_KEY'), openai_api_base='https://api.deepseek.com', max_tokens=1024)# 3️⃣ 自定义 Agent：支持 AgentFinish 输出class SmartLLMAgent(BaseSingleActionAgent): llm: ChatOpenAI tools: Sequence[Tool] @property def input_keys(self) -> List[str]: return ["input"] def plan( self, intermediate_steps: List[Tuple[AgentAction, str]], **kwargs: Any, ) -> AgentAction | AgentFinish: user_input = kwargs["input"] # 构造 scratchpad scratchpad = "" for i, (action, obs) in enumerate(intermediate_steps): scratchpad += f"第{i + 1}步：我调用了 {action.tool}，输入是「{action.tool_input}」，得到结果「{obs}」。\n" print("中间步骤记事本：", scratchpad) prompt = [ SystemMessage(content=( "你是一个智能代理。你可以使用以下工具：" f"{', '.join([tool.name for tool in self.tools])}。\n" "你的目标是根据用户输入合理地调用工具，或者在觉得任务完成时输出最终答案。\n" "请根据格式返回：\n" "调用工具：TOOL_NAME | TOOL_INPUT\n" "给出最终答案：FINAL_ANSWER | RESULT\n" )), HumanMessage(content=( f"用户的问题是：{user_input}\n\n" f"当前中间步骤如下：\n{scratchpad if scratchpad else '（尚未调用工具）'}" )) ] output = self.llm.invoke(prompt).content.strip() print("[LLM 输出]：", output) if "给出最终答案" in output: output = output.split("：")[1] final_answer = output.split("|", 1)[1].strip() return AgentFinish(return_values={"output": final_answer}, log=output) elif "调用工具" in output: output = output.split("：")[1] tool, tool_input = [x.strip() for x in output.split("|", 1)] return AgentAction(tool=tool, tool_input=tool_input, log=output) else: raise ValueError("无法解析 LLM 输出格式") async def aplan( self, intermediate_steps: list[tuple[AgentAction, str]], **kwargs: Any, ) -> AgentAction | AgentFinish: pass# 4️⃣ 构建 AgentExecutoragent = SmartLLMAgent(llm=llm, tools=tools)executor = AgentExecutor( agent=agent, tools=tools, verbose=False, handle_parsing_errors=True)# 5️⃣ 测试运行result = executor.invoke({"input": "请把LangChain倒过来并告诉我结果"})print("\n最终结果：", result["output"])四、调整策略与中间态管理建议
- 可自定义插件延迟缓存工具输出、过滤结果、整合多个工具可读取
agent_executor.intermediate_steps 来保存中间态日志或批量存储当工具较多时，建议对 tools 列表进行命名和分类管理agent_executor.return_intermediate_steps = Trueresult = agent_executor.invoke( {"input": "请你优先使用我提供的工具反转字符串：LangChain"})from pprint import pprintpprint(result)小结
- 自定义
BaseTool 提供了灵活的工具功能扩展能力；插件机制（callbacks）让你可以插入日志、缓存、限制等额外逻辑；AgentAction 与 AgentFinish 是控制 Agent 推理循环的标准结构，使你可以更可控地管理中间态与终止。接下来我们将深入拆解 RetrievalQA 的内部架构，探究如何实现检索增强生成（RAG）。
