如何强制工具调用代理生成结构化输出¶
您可能希望您的代理返回结构化的输出。例如，如果代理的输出被其他下游软件使用，您可能希望每次调用代理时输出都保持相同的结构化格式，以确保一致性。
本笔记本将介绍两种不同的方法，用于强制工具调用代理生成结构化输出。我们将使用一个基本的ReAct代理（一个模型节点和一个工具调用节点）以及一个用于格式化响应的第三个节点。这两种选项将使用相同的图结构，如下面的图所示，但内部机制不同。
选项 1
强制工具调用代理生成结构化输出的第一种方法是将您希望的输出绑定为代理节点使用的附加工具。与基本的 ReAct 代理不同，这里的代理节点不是在 tools 和 END 之间选择，而是在它调用的具体工具之间选择。在这种情况下，预期的流程是 LLM 在代理节点中首先选择动作工具，然后在收到动作工具输出后调用响应工具，响应工具将路由到 respond 节点，该节点只是从代理节点工具调用中结构化参数。
优缺点
这种格式的优点是您只需要一个 LLM，并且由于这一点可以节省成本和延迟。这种方法的缺点是不能保证单个 LLM 在您希望它调用时会调用正确的工具。我们可以通过将 tool_choice 设置为 any 来帮助 LLM，当使用 bind_tools 时这将强制 LLM 在每一步至少选择一个工具，但这远非万无一失的策略。此外，另一个缺点是代理可能会调用多个工具，因此我们需要在路由函数中显式检查这一点（或者如果我们使用 OpenAI，可以将 parallell_tool_calling=False 设置为确保一次只调用一个工具）。
选项 2
强制工具调用代理生成结构化输出的第二种方法是使用第二个 LLM（在这种情况下为 model_with_structured_output）来响应用户。
在这种情况下，您将定义一个基本的 ReAct 代理，但不是让代理节点在工具节点和结束对话之间选择，而是让代理节点在工具节点和响应节点之间选择。响应节点将包含一个使用结构化输出的第二个 LLM，一旦被调用将直接返回给用户。您可以将这种方法视为基本 ReAct 代理在响应用户之前多加一步。
优缺点
这种方法的优点是它保证了结构化输出（只要 .with_structured_output 与 LLM 的预期一致）。这种方法的缺点是它需要在响应用户之前进行额外的 LLM 调用，这会增加成本以及延迟。此外，由于没有向代理节点 LLM 提供关于所需输出模式的信息，存在代理 LLM 未能调用正确工具以回答正确输出模式的风险。
请注意，这两种选项将遵循完全相同的图结构（参见上面的图），即它们都是基本 ReAct 架构的精确副本，但在结束之前有一个响应节点。
设置¶
首先，让我们安装所需的包并设置我们的 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 方法，您可以在此处了解更多相关信息 here。
在这个示例中，我们将使用一个工具来查找天气，并返回一个结构化的天气响应给用户。
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)
API Reference: tool | ChatAnthropic
选项 1：将输出绑定为工具¶
现在让我们看看如何使用单个LLM选项。
定义图¶
图的定义与上面的非常相似，唯一的区别是我们不再在 response 节点中调用LLM，而是将 WeatherResponse 工具绑定到已经包含 get_weather 工具的LLM上。
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()
API Reference: StateGraph | END | ToolNode
使用方法¶
现在我们可以运行我们的图来检查它是否按预期工作：
answer = graph.invoke(input={"messages": [("human", "what's the weather in SF?")]})[
"final_response"
]
再次，代理返回了我们期望的WeatherResponse对象。
选项 2：两个 LLM¶
现在让我们来看看如何使用第二个 LLM 来强制生成结构化的输出。
定义图¶
我们现在可以定义我们的图：
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()
API Reference: StateGraph | END | ToolNode | HumanMessage
使用方法¶
我们现在可以调用我们的图来验证输出是否如预期那样被结构化：
answer = graph.invoke(input={"messages": [("human", "what's the weather in SF?")]})[
"final_response"
]
如我们所见，代理返回了一个WeatherResponse对象，符合我们的预期。现在，可以轻松地在更复杂的软件堆栈中使用此代理，而无需担心代理的输出与堆栈中下一步所期望的格式不匹配。