29. LangChain——Agent的middleware中间件
本帖最后由 糖逗 于 2026-3-9 16:01 编辑1. 中间件的作用？
中间件的作用是对智能体的每一步工作进行控制和自定义的执行。
LangChain中内置了一些基础的中间件：https://docs.langchain.com/oss/python/langchain/middlewarre/built-in
2. 中间件可以嵌入在哪些节点中？
3.代码实战
from typing import Dict, Any, List
from uuid import UUID
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.outputs import LLMResult
# ================= 2. 定义工具 =================
@tool(description="查询天气，传入城市名称字符串，返回字符串天气信息")
def get_weather(city: str) -> str:
return f"{city}天气：晴天"
# ================= 3. 【核心修改】自定义回调处理器 (替代旧版 Middleware) =================
class MyCustomHandler(BaseCallbackHandler):
"""
自定义回调类，通过重写以下方法来实现 before/after 的逻辑
"""
# 对应: Agent 启动
def on_chain_start(self, serialized: Dict, inputs: Dict, **kwargs: Any) -> Any:
print(f" agent启动，并附带消息")
# 对应: Agent 结束
def on_chain_end(self, outputs: Dict, **kwargs: Any) -> Any:
print(f" agent结束")
# 对应: Model 执行前
def on_llm_start(self, serialized: Dict, prompts: List, **kwargs: Any) -> Any:
print(f" 模型即将调用")
print("模型调用啦")
# 对应: Model 执行后
def on_llm_end(self, response: LLMResult, **kwargs: Any) -> Any:
print(f" 模型调用结束")
# 对应: Tool 执行前
def on_tool_start(self, serialized: Dict, input_str: str, **kwargs: Any) -> Any:
tool_name = kwargs.get('name', 'unknown_tool')
print(f"工具执行：{tool_name}")
print(f"工具执行传入参数：{input_str}")
# ================= 4. 初始化模型和 Agent =================
llm = ChatOpenAI(
model="qwen-max",
temperature=0,
openai_api_base="https://dashscope.aliyuncs.com/compatible-mode/v1",
)
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个天气助手。"),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad"),
])
agent = create_tool_calling_agent(llm=llm, tools=, prompt=prompt)
agent_executor = AgentExecutor(agent=agent, tools=, verbose=False)
# ================= 5. 执行并传入回调 =================
# 在这里把我们写好的 Handler 传进去
my_handler = MyCustomHandler()
print("--- 开始运行 ---")
res = agent_executor.invoke(
{"input": "深圳今天的天气如何呀，如何穿衣"},
config={"callbacks": }
)
print("\n**********\n最终结果:", res["output"])
中间件的作用？
中间件的作用是对智能体的每一步工作进行控制和自定义的执行。
在LangChain中，中间件（现在通常通过回调处理器实现）允许我们在智能体执行的各个阶段插入自定义逻辑，比如日志记录、监控、修改输入输出等。
中间件可以嵌入在哪些节点中？
通过自定义回调处理器，我们可以嵌入以下节点（事件）：
[*]链（Chain）开始和结束
[*]大型语言模型（LLM）调用开始和结束
[*]工具（Tool）调用开始和结束
[*]其他如检索、提示等事件
代码实战
以下是修复后的完整代码示例：
from typing import Dict, Any, List
from uuid import UUID
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.outputs import LLMResult
# ================= 1. 定义工具 =================
@tool(description="查询天气，传入城市名称字符串，返回字符串天气信息")
def get_weather(city: str) -> str:
return f"{city}天气：晴天"
# ================= 2. 自定义回调处理器 =================
class MyCustomHandler(BaseCallbackHandler):
"""
自定义回调类，通过重写方法实现中间件逻辑
"""
def on_chain_start(self, serialized: Dict, inputs: Dict, **kwargs: Any) -> Any:
print(f" agent启动，并附带消息")
def on_chain_end(self, outputs: Dict, **kwargs: Any) -> Any:
print(f" agent结束")
def on_llm_start(self, serialized: Dict, prompts: List, **kwargs: Any) -> Any:
print(f" 模型即将调用")
print("模型调用啦")
def on_llm_end(self, response: LLMResult, **kwargs: Any) -> Any:
print(f" 模型调用结束")
def on_tool_start(self, serialized: Dict, input_str: str, **kwargs: Any) -> Any:
tool_name = kwargs.get('name', 'unknown_tool')
print(f"工具执行：{tool_name}")
print(f"工具执行传入参数：{input_str}")
# ================= 3. 初始化模型和Agent =================
llm = ChatOpenAI(
model="qwen-max",
temperature=0,
openai_api_base="https://dashscope.aliyuncs.com/compatible-mode/v1",
)
# 构建提示模板
prompt = ChatPromptTemplate.from_messages([
("system", "You are a helpful assistant."),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad"),
])
# 工具列表
tools =
agent = create_tool_calling_agent(llm=llm, tools=tools, prompt=prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=False)
# ================= 4. 执行并传入回调处理器 =================
my_handler = MyCustomHandler()
print("--- 开始运行 ---")
res = agent_executor.invoke(
{"input": "深圳今天的天气如何呀，如何穿衣"},
config={"callbacks": }
)
print("\n**********\n最终结果:", res)
关键点说明：
[*]我们创建了一个自定义回调处理器 MyCustomHandler，通过重写特定方法（如 on_chain_start、on_tool_start 等）来在智能体执行的不同阶段添加自定义操作。
[*]在初始化智能体执行器后，通过 config={"callbacks": } 将回调处理器传入，从而实现对执行过程的监控和控制。
运行此代码，你将看到在智能体执行过程中，各个节点（如链启动、模型调用、工具执行等）都会触发我们定义的回调方法，打印相应的信息。这就是LangChain中利用回调处理器实现中间件功能的方式。
以上回复来自 FishC AI 小助理，如未能正确解答您的问题，请继续追问。
页:
[1]