如何强制工具调用代理生成结构化输出¶
您可能希望您的代理返回结构化的输出。例如，如果代理的输出被下游软件使用，您可能希望每次调用代理时输出都保持相同的结构化格式以确保一致性。
此笔记本将介绍两种不同的方法来强制工具调用代理生成结构化输出。我们将使用一个基本的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项目中的问题，并提高其性能。通过使用跟踪数据，您可以调试、测试和监控使用LangGraph构建的LLM应用程序——更多关于如何开始的信息，请参阅这里。
定义模型、工具和图状态¶
现在我们可以定义我们希望如何构建输出结构，定义我们的图状态，并且定义我们将要使用的工具和模型。
为了使用结构化输出，我们将使用LangChain中的with_structured_output方法，更多详情可以参考这里。
在这个示例中，我们将使用一个工具来查找天气，并返回一个结构化的天气响应给用户。
API Reference: tool
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，而是将包含get_weather工具的LLM绑定到WeatherResponse工具上。
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：两个大语言模型¶
现在让我们来看看如何使用第二个大语言模型来强制生成结构化的输出。
定义图¶
我们现在可以定义我们的图：
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对象，符合我们的预期。现在可以很容易地在更复杂的软件堆栈中使用这个代理，并且无需担心代理的输出与堆栈中下一个步骤期望的格式不匹配。