OpenAI RAG+LangChain工具调用与withStructuredOutput兼容问题排查

一、工具触发失败的排查与解决 #

1. 优化工具定义的描述精度 #

LLM无法触发工具的核心原因通常是工具描述不够明确，未覆盖所有触发场景。需给talkToAgentTool补充精准的触发规则：

from langchain.tools import Tool

def talk_to_agent(input: str) -> str:
    return "正在为您转接人工客服，请稍候..."

talkToAgentTool = Tool(
    name="talk_to_agent",
    func=talk_to_agent,
    description="当用户明确要求联系人工客服时必须调用，触发关键词包括：'agent'、'talk to human'、'talk to agent'、'转人工'等"
)

2. 强化Prompt的引导规则 #

在系统Prompt中明确告知LLM触发工具的条件，避免其忽略用户的转人工请求：

from langchain.prompts import ChatPromptTemplate

system_prompt = """
你是基于RAG的智能助手，优先使用检索到的信息回答问题。
若用户输入包含'talk to human'、'agent'、'talk to agent'等转人工请求，**必须调用talk_to_agent工具**，禁止自行回答。
"""
prompt = ChatPromptTemplate.from_messages([
    ("system", system_prompt),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}")
])

3. 兼容withStructuredOutput的处理 #

如果使用了结构化输出约束，需在输出结构中预留工具调用字段，避免限制LLM的工具触发能力：

from langchain_core.output_parsers import JsonOutputParser
from pydantic import BaseModel, Field

# 定义包含工具调用字段的结构化输出模型
class ResponseSchema(BaseModel):
    answer: str = Field(description="正常回答内容，转人工时留空")
    tool_call: str = Field(description="需调用的工具名，转人工时填'talk_to_agent'，否则留空")

parser = JsonOutputParser(pydantic_object=ResponseSchema)

# 更新Prompt加入结构化输出规则
system_prompt = """
你是基于RAG的智能助手，优先使用检索到的信息回答问题。
若用户请求转人工，必须在tool_call字段填写'talk_to_agent'，answer字段留空；否则在answer字段填写回答内容，tool_call字段留空。
输出必须符合以下格式：
{format_instructions}
"""
prompt = ChatPromptTemplate.from_messages([
    ("system", system_prompt.format(format_instructions=parser.get_format_instructions())),
    ("human", "{input}")
])

# 处理输出的逻辑
def process_llm_response(response):
    parsed_data = parser.parse(response.content)
    if parsed_data.tool_call == "talk_to_agent":
        return talk_to_agent("")
    return parsed_data.answer

二、调用工具后终止对话流程的实现 #

1. 异常抛出法 #

在工具函数中抛出特定异常，对话循环捕获后直接终止流程：

# 定义自定义异常类
class TransferToHumanError(Exception):
    pass

def talk_to_agent(input: str) -> str:
    # 触发转接逻辑后抛出异常
    raise TransferToHumanError("已触发人工客服转接，当前对话终止")

# 对话执行逻辑
def run_chat_session(user_input):
    try:
        # 执行Agent或RAG逻辑
        result = agent_executor.invoke({"input": user_input})
        return result["output"]
    except TransferToHumanError as e:
        return str(e)

2. 结果检查法 #

在Agent执行结果中检查是否调用了目标工具，若已调用则直接返回结果并终止后续流程：

from langchain.agents import AgentExecutor, create_openai_tools_agent

# 创建Agent及执行器
agent = create_openai_tools_agent(llm, [talkToAgentTool], prompt)
agent_executor = AgentExecutor(
    agent=agent,
    tools=[talkToAgentTool],
    handle_parsing_errors=True,
    return_intermediate_steps=True
)

# 处理对话结果
def execute_chat(user_input):
    response = agent_executor.invoke({"input": user_input})
    # 检查中间步骤是否调用了talk_to_agent
    for step in response.get("intermediate_steps", []):
        if step[0].tool == "talk_to_agent":
            return step[1]
    return response["output"]

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