如何强制工具调用代理结构化输出¶
您可能希望您的代理以结构化格式返回输出。例如，如果代理的输出被其他下游软件使用，您可能希望每次调用代理时都使用相同的结构化格式以确保一致性。
本笔记本将介绍两种不同的方法来强制工具调用代理进行结构化输出。我们将使用一个基本的 ReAct 代理（一个模型节点和一个工具调用节点），并在最后添加第三个节点，用于为用户格式化响应。这两种方法将使用与下图中所示相同的图结构，但其内部机制有所不同。
选项 1
强制工具调用代理具有结构化输出的第一种方式是将您期望的输出作为附加工具绑定到 agent 节点上使用。与基本的 ReAct 代理不同，这种情况下 agent 节点不是在 tools 和 END 之间选择，而是在特定的工具之间选择。在这种情况下，预期的流程是 agent 节点中的 LLM 首先选择操作工具，然后在接收到操作工具的输出后调用响应工具，该工具随后将路由到 respond 节点，该节点仅对 agent 节点工具调用的参数进行结构化。
优缺点
这种格式的优点是只需要一个 LLM，可以节省成本和延迟。这种方法的缺点是不能保证单个 LLM 在您需要的时候调用正确的工具。我们可以通过在使用 bind_tools 时将 tool_choice 设置为 any 来帮助 LLM，这迫使 LLM 在每一步至少选择一个工具，但这远非万无一失的策略。此外，另一个缺点是代理可能会调用 多个 工具，因此我们需要在路由函数中显式检查这一点（或者如果我们使用的是 OpenAI，我们可以设置 parallell_tool_calling=False 来确保一次只调用一个工具）。
选项 2
强制工具调用代理具有结构化输出的第二种方式是使用第二个 LLM（在这种情况下是 model_with_structured_output）来向用户响应。
在这种情况下，您将正常定义一个基本的 ReAct 代理，但是 agent 节点不再在 tools 节点和结束对话之间进行选择，而是会在 tools 节点和 respond 节点之间进行选择。respond 节点将包含一个使用结构化输出的第二个 LLM，一旦被调用，它将直接返回给用户。您可以将此方法视为基本的 ReAct，在向用户响应之前增加了一个额外的步骤。
优缺点
这种方法的优点是只要 .with_structured_output 按预期与 LLM 一起工作，就可以保证结构化输出。这种方法的缺点是需要在向用户响应之前进行额外的 LLM 调用，这会增加成本和延迟。此外，由于没有向 agent 节点的 LLM 提供所需输出模式的信息，存在 agent 的 LLM 可能无法调用正确工具以按照正确的输出模式回答的风险。
请注意，这两种选项都将遵循完全相同的图结构（参见上面的图表），因为它们都是基本 ReAct 架构的精确复制品，只是在结束前有一个 respond 节点。
设置¶
首先，让我们安装所需的包并设置我们的 API 密钥
import getpass
import os
def _set_env(var: str):
if not os.environ.get(var):
os.environ[var] = getpass.getpass(f"{var}: ")
_set_env("ANTHROPIC_API_KEY")
为 LangGraph 开发设置 LangSmith
注册 LangSmith 可以快速发现并改进你的 LangGraph 项目的问题。LangSmith 允许你使用跟踪数据来调试、测试和监控使用 LangGraph 构建的 LLM 应用 —— 有关如何入门的更多信息，请阅读 此处。
定义模型、工具和图状态¶
现在我们可以定义我们希望如何构建输出，定义我们的图状态，以及我们将使用的工具和模型。
为了使用结构化输出，我们将使用 LangChain 的 with_structured_output 方法，你可以在 这里 了解更多相关内容。
在这个示例中，我们将使用一个用于查找天气的工具，并向用户返回结构化的天气响应。
API Reference: tool | ChatAnthropic
from pydantic import BaseModel, Field
from typing import Literal
from langchain_core.tools import tool
from langchain_anthropic import ChatAnthropic
from langgraph.graph import MessagesState
class WeatherResponse(BaseModel):
"""Respond to the user with this"""
temperature: float = Field(description="The temperature in fahrenheit")
wind_directon: str = Field(
description="The direction of the wind in abbreviated form"
)
wind_speed: float = Field(description="The speed of the wind in km/h")
# Inherit 'messages' key from MessagesState, which is a list of chat messages
class AgentState(MessagesState):
# Final structured response from the agent
final_response: WeatherResponse
@tool
def get_weather(city: Literal["nyc", "sf"]):
"""Use this to get weather information."""
if city == "nyc":
return "It is cloudy in NYC, with 5 mph winds in the North-East direction and a temperature of 70 degrees"
elif city == "sf":
return "It is 75 degrees and sunny in SF, with 3 mph winds in the South-East direction"
else:
raise AssertionError("Unknown city")
tools = [get_weather]
model = ChatAnthropic(model="claude-3-opus-20240229")
model_with_tools = model.bind_tools(tools)
model_with_structured_output = model.with_structured_output(WeatherResponse)
选项 1：将输出绑定为工具¶
现在让我们看看如何使用单个 LLM 选项。
定义图¶
该图的定义与上面的非常相似，唯一的不同是我们不再在 response 节点中调用 LLM，而是将 WeatherResponse 工具绑定到我们已经包含 get_weather 工具的 LLM 上。
API Reference: StateGraph | END | ToolNode
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
tools = [get_weather, WeatherResponse]
# Force the model to use tools by passing tool_choice="any"
model_with_response_tool = model.bind_tools(tools, tool_choice="any")
# Define the function that calls the model
def call_model(state: AgentState):
response = model_with_response_tool.invoke(state["messages"])
# We return a list, because this will get added to the existing list
return {"messages": [response]}
# Define the function that responds to the user
def respond(state: AgentState):
# Construct the final answer from the arguments of the last tool call
weather_tool_call = state["messages"][-1].tool_calls[0]
response = WeatherResponse(**weather_tool_call["args"])
# Since we're using tool calling to return structured output,
# we need to add a tool message corresponding to the WeatherResponse tool call,
# This is due to LLM providers' requirement that AI messages with tool calls
# need to be followed by a tool message for each tool call
tool_message = {
"type": "tool",
"content": "Here is your structured response",
"tool_call_id": weather_tool_call["id"],
}
# We return the final answer
return {"final_response": response, "messages": [tool_message]}
# Define the function that determines whether to continue or not
def should_continue(state: AgentState):
messages = state["messages"]
last_message = messages[-1]
# If there is only one tool call and it is the response tool call we respond to the user
if (
len(last_message.tool_calls) == 1
and last_message.tool_calls[0]["name"] == "WeatherResponse"
):
return "respond"
# Otherwise we will use the tool node again
else:
return "continue"
# Define a new graph
workflow = StateGraph(AgentState)
# Define the two nodes we will cycle between
workflow.add_node("agent", call_model)
workflow.add_node("respond", respond)
workflow.add_node("tools", ToolNode(tools))
# Set the entrypoint as `agent`
# This means that this node is the first one called
workflow.set_entry_point("agent")
# We now add a conditional edge
workflow.add_conditional_edges(
"agent",
should_continue,
{
"continue": "tools",
"respond": "respond",
},
)
workflow.add_edge("tools", "agent")
workflow.add_edge("respond", END)
graph = workflow.compile()
用法¶
现在我们可以运行我们的图，以检查它是否按预期工作：
answer = graph.invoke(input={"messages": [("human", "what's the weather in SF?")]})[
"final_response"
]
同样，代理返回了一个我们预期的 WeatherResponse 对象。
选项 2：使用 2 个 LLM¶
现在让我们深入探讨如何使用第二个 LLM 来强制生成结构化输出。
定义图¶
我们现在可以定义我们的图：
API Reference: StateGraph | END | ToolNode | HumanMessage
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_core.messages import HumanMessage
# Define the function that calls the model
def call_model(state: AgentState):
response = model_with_tools.invoke(state["messages"])
# We return a list, because this will get added to the existing list
return {"messages": [response]}
# Define the function that responds to the user
def respond(state: AgentState):
# We call the model with structured output in order to return the same format to the user every time
# state['messages'][-2] is the last ToolMessage in the convo, which we convert to a HumanMessage for the model to use
# We could also pass the entire chat history, but this saves tokens since all we care to structure is the output of the tool
response = model_with_structured_output.invoke(
[HumanMessage(content=state["messages"][-2].content)]
)
# We return the final answer
return {"final_response": response}
# Define the function that determines whether to continue or not
def should_continue(state: AgentState):
messages = state["messages"]
last_message = messages[-1]
# If there is no function call, then we respond to the user
if not last_message.tool_calls:
return "respond"
# Otherwise if there is, we continue
else:
return "continue"
# Define a new graph
workflow = StateGraph(AgentState)
# Define the two nodes we will cycle between
workflow.add_node("agent", call_model)
workflow.add_node("respond", respond)
workflow.add_node("tools", ToolNode(tools))
# Set the entrypoint as `agent`
# This means that this node is the first one called
workflow.set_entry_point("agent")
# We now add a conditional edge
workflow.add_conditional_edges(
"agent",
should_continue,
{
"continue": "tools",
"respond": "respond",
},
)
workflow.add_edge("tools", "agent")
workflow.add_edge("respond", END)
graph = workflow.compile()
用法¶
我们现在可以调用我们的图以验证输出是否按预期进行结构化：
answer = graph.invoke(input={"messages": [("human", "what's the weather in SF?")]})[
"final_response"
]
正如我们所看到的，代理返回了一个我们预期的 WeatherResponse 对象。现在，我们很容易在更复杂的软件栈中使用这个代理，而无需担心代理的输出与软件栈下一步骤所期望的格式不匹配。