Skip to content

如何强制工具调用代理输出结构化结果

你可能希望你的代理以结构化格式返回其输出。例如,如果代理的输出将被其他下游软件使用,你可能希望每次调用代理时输出都采用相同的结构化格式,以确保一致性。

本笔记本将介绍两种强制工具调用代理输出结构化结果的不同方法。我们将使用一个基本的 ReAct 代理(一个模型节点和一个工具调用节点),并在最后添加一个第三节点,该节点将为用户格式化响应。这两种方法都将使用如下图所示的相同图结构,但底层机制不同。

react_diagrams.png

方法一

option1.png

强制工具调用代理输出结构化结果的第一种方法是,将你期望的输出作为一个额外的工具绑定给 agent 节点使用。与基本的 ReAct 代理不同,在这种情况下,agent 节点不是在 toolsEND 之间进行选择,而是在它调用的特定工具之间进行选择。在这种情况下,预期的流程是,agent 节点中的大语言模型(LLM)将首先选择操作工具,在收到操作工具的输出后,它将调用响应工具,然后该工具将路由到 respond 节点,该节点只是对 agent 节点工具调用的参数进行结构化处理。

优缺点

这种格式的优点是你只需要一个大语言模型,因此可以节省成本和减少延迟。这种方法的缺点是,不能保证单个大语言模型会在你需要的时候调用正确的工具。当我们使用 bind_tools 时,将 tool_choice 设置为 any 可以帮助大语言模型,这会强制大语言模型在每一轮至少选择一个工具,但这远不是一个万无一失的策略。此外,另一个缺点是代理可能会调用 多个 工具,因此我们需要在路由函数中明确检查这一点(或者如果我们使用的是 OpenAI,可以将 parallell_tool_calling=False 设置为确保每次只调用一个工具)。

方法二

option2.png

强制工具调用代理输出结构化结果的第二种方法是使用第二个大语言模型(在这种情况下是 model_with_structured_output)来响应用户。

在这种情况下,你通常会定义一个基本的 ReAct 代理,但 agent 节点不是在 tools 节点和结束对话之间进行选择,而是在 tools 节点和 respond 节点之间进行选择。respond 节点将包含一个使用结构化输出的第二个大语言模型,一旦被调用,它将直接返回给用户。你可以将这种方法看作是基本的 ReAct 方法,只是在响应用户之前多了一步。

优缺点

这种方法的优点是它保证了结构化输出(只要 .with_structured_output 与大语言模型配合使用时能按预期工作)。使用这种方法的缺点是,在响应用户之前需要额外调用一次大语言模型,这会增加成本和延迟。此外,由于没有向 agent 节点的大语言模型提供有关所需输出模式的信息,存在 agent 大语言模型无法调用以正确输出模式回答问题所需的正确工具的风险。

请注意,这两种方法都将遵循完全相同的图结构(见上图),因为它们都是基本 ReAct 架构的精确复制品,只是在结尾前有一个 respond 节点。

环境设置

首先,让我们安装所需的包并设置我们的 API 密钥

%%capture --no-stderr
%pip install -U langgraph langchain_anthropic
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 构建的大语言模型应用程序 —— 点击 此处 了解更多关于如何开始使用的信息。

定义模型、工具和图状态

现在我们可以定义我们希望如何构建输出结构、定义图状态,以及我们要使用的工具和模型。

为了使用结构化输出,我们将使用 LangChain 的 with_structured_output 方法,你可以在此处了解更多相关信息。

在这个示例中,我们将使用一个用于查询天气的工具,并向用户返回结构化的天气响应。

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

选项 1:将输出绑定为工具

现在让我们来研究一下如何使用单一大语言模型(LLM)选项。

定义图

图的定义与上面的非常相似,唯一的区别是我们不再在 response 节点中调用大语言模型,而是将 WeatherResponse 工具绑定到已经包含 get_weather 工具的大语言模型上。

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"
]
answer
WeatherResponse(temperature=75.0, wind_directon='SE', wind_speed=3.0)

同样,代理如我们所预期的那样返回了一个 WeatherResponse 对象。

选项 2:使用两个大语言模型(LLMs)

现在让我们深入探讨如何使用第二个大语言模型来强制生成结构化输出。

定义图

现在我们可以定义我们的图:

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: HumanMessage

使用方法

现在,我们可以调用我们的图来验证输出是否按预期进行结构化:

answer = graph.invoke(input={"messages": [("human", "what's the weather in SF?")]})[
    "final_response"
]
answer
WeatherResponse(temperature=75.0, wind_directon='SE', wind_speed=4.83)

正如我们所见,该智能体如我们预期的那样返回了一个 WeatherResponse 对象。现在,在更复杂的软件栈中使用这个智能体将变得很容易,无需担心智能体的输出与软件栈下一步所期望的格式不匹配。

Comments