Langchain中通过AgentExecutor向工具传递自定义运行时参数的问题

核心问题原因 #

使用InjectedToolArg标记的字段会被LangChain排除在LLM生成范围外，必须通过运行时主动注入才能通过Pydantic校验。之前的两种方法失败，是因为参数注入的位置或方式不符合v0.3版本的工具调用逻辑。

正确实现步骤 #

1. 规范定义工具输入模型与工具

确保工具输入模型继承HumanInTheLoopInput，并用InjectedToolArg标记需要注入的字段，工具函数正确接收该参数：

from langchain.tools import HumanInTheLoopInput, InjectedToolArg, AsyncTool
from pydantic import BaseModel

class CustomHITLInput(HumanInTheLoopInput):
    conversation_id: str = InjectedToolArg(description="对话唯一标识ID")

async def hitl_tool(input_data: CustomHITLInput) -> str:
    # 这里编写工具逻辑，使用传入的conversation_id
    return f"已处理对话ID: {input_data.conversation_id}"

# 实例化工具
hitl_tool_instance = AsyncTool.from_function(
    func=hitl_tool,
    name="human_in_the_loop",
    description="用于人工干预的工具，conversation_id由系统自动注入"
)

2. 通过回调处理器注入参数（可靠方案）

自定义AsyncCallbackHandler，在工具调用启动时修改工具输入，注入conversation_id：

from langchain.callbacks.base import AsyncCallbackHandler
import json
from typing import Dict, Any

class HITLParamInjector(AsyncCallbackHandler):
    def __init__(self, conversation_id: str):
        self.conversation_id = conversation_id

    async def on_tool_start(
        self,
        serialized: Dict[str, Any],
        input_str: str,
        **kwargs: Any
    ) -> None:
        # 解析原工具输入为字典
        tool_input = json.loads(input_str)
        # 注入自定义参数
        tool_input["conversation_id"] = self.conversation_id
        # 返回修改后的输入字符串，确保工具接收正确参数
        modified_input_str = json.dumps(tool_input)
        return super().on_tool_start(serialized, modified_input_str, **kwargs)

3. 调用AgentExecutor时绑定回调

创建回调实例并传入AgentExecutor，确保参数在工具调用时被注入：

# 假设已完成Agent的构建（如StructuredChatAgent）
agent_executor = AgentExecutor(agent=agent, tools=[hitl_tool_instance])

# 传入目标对话ID
target_conv_id = "conv_20240520_001"
callback = HITLParamInjector(target_conv_id)

# 调用工具，参数将通过回调注入
result = await agent_executor.ainvoke(
    {"input": "需要人工处理的请求内容"},
    callbacks=[callback]
)

4. 修正Agent提示词

在Agent的系统提示词中明确告知LLM不需要生成conversation_id，避免混淆：

from langchain.prompts import ChatPromptTemplate

system_prompt = """
你是任务处理助手，当调用human_in_the_loop工具时，无需生成conversation_id参数，该参数由系统自动注入。
仅需根据用户需求填写工具的其他必要参数。
"""

prompt = ChatPromptTemplate.from_messages([
    ("system", system_prompt),
    ("human", "{input}"),
    ("ai", "{agent_scratchpad}")
])

验证与排查 #

• 在工具函数中添加打印语句，确认conversation_id是否被正确接收：

  async def hitl_tool(input_data: CustomHITLInput) -> str:
      print(f"实际接收的conversation_id: {input_data.conversation_id}")
      return f"已处理对话ID: {input_data.conversation_id}"
  

• 若仍出现校验错误，检查工具输入模型的字段名是否与注入的参数名完全一致，确保无拼写错误。

内容的提问来源于stack exchange，提问作者Thomas Dussaut