29. LangChain——middleware中间件

糖逗 · 发表于 2 小时前

马上注册，结交更多好友，享用更多功能^_^

您需要登录才可以下载或查看，没有账号？立即注册

x

1. 中间件的作用？
中间件的作用是对智能体的每一步工作进行控制和自定义的执行。
下载 (51).png

LangChain中内置了一些基础的中间件：https://docs.langchain.com/oss/python/langchain/middlewarre/built-in

2. 中间件可以嵌入在哪些节点中？
下载 (52).png

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[str, Any], inputs: Dict[str, Any], **kwargs: Any) -> Any:
print(f"[before agent] agent启动，并附带消息")
# 对应: Agent 结束
def on_chain_end(self, outputs: Dict[str, Any], **kwargs: Any) -> Any:
print(f"[after agent] agent结束")
# 对应: Model 执行前
def on_llm_start(self, serialized: Dict[str, Any], prompts: List[str], **kwargs: Any) -> Any:
print(f"[before_model] 模型即将调用")
print("模型调用啦")
# 对应: Model 执行后
def on_llm_end(self, response: LLMResult, **kwargs: Any) -> Any:
print(f"[after_model] 模型调用结束")
# 对应: Tool 执行前
def on_tool_start(self, serialized: Dict[str, Any], 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=[get_weather], prompt=prompt)
agent_executor = AgentExecutor(agent=agent, tools=[get_weather], verbose=False)
# ================= 5. 执行并传入回调 =================
# 在这里把我们写好的 Handler 传进去
my_handler = MyCustomHandler()
print("--- 开始运行 ---")
res = agent_executor.invoke(
{"input": "深圳今天的天气如何呀，如何穿衣"},
config={"callbacks": [my_handler]}
)
print("\n**********\n最终结果:", res["output"])

复制代码

FishC · 发表于 2 小时前

中间件的作用？
中间件的作用是对智能体的每一步工作进行控制和自定义的执行。

在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="[url=https://dashscope.aliyuncs.com/compatible-mode/v1",]https://dashscope.aliyuncs.com/compatible-mode/v1",[/url]
)
# 构建提示模板
prompt = ChatPromptTemplate.from_messages([
("system", "You are a helpful assistant."),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad"),
])
# 工具列表
tools = [get_weather]
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": [my_handler]}
)
print("\n**********\n最终结果:", res)

复制代码

关键点说明：

我们创建了一个自定义回调处理器 MyCustomHandler，通过重写特定方法（如 on_chain_start、on_tool_start 等）来在智能体执行的不同阶段添加自定义操作。
在初始化智能体执行器后，通过 config={"callbacks": [my_handler]} 将回调处理器传入，从而实现对执行过程的监控和控制。

运行此代码，你将看到在智能体执行过程中，各个节点（如链启动、模型调用、工具执行等）都会触发我们定义的回调方法，打印相应的信息。这就是LangChain中利用回调处理器实现中间件功能的方式。

以上回复来自 FishC AI 小助理，如未能正确解答您的问题，请继续追问。

账号		自动登录	找回密码
密码			立即注册

[AI工作流] 29. LangChain——middleware中间件

马上注册，结交更多好友，享用更多功能^_^

本帖被以下淘专辑推荐:

浏览过的版块