如何使用 ReAct 风格的代理返回结构化输出¶

%%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")

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):
    """你接收到的数据训练到2023年10月。"""

    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_response: WeatherResponse


@tool
def get_weather(city: Literal["nyc", "sf"]):
    """使用此信息获取天气信息。"""
    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)

from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode

tools = [get_weather, WeatherResponse]

# 通过传递 tool_choice= 强制模型使用工具。"any"
model_with_response_tool = model.bind_tools(tools, tool_choice="any")


# 定义调用模型的函数。
def call_model(state: AgentState):
    response = model_with_response_tool.invoke(state["messages"])
    # 我们返回一个列表，因为这将被添加到现有列表中。
    return {"messages": [response]}


# 定义响应用户的函数。
def respond(state: AgentState):
    # 根据最后一次工具调用的参数构建最终答案。
    response = WeatherResponse(**state["messages"][-1].tool_calls[0]["args"])
    # 我们返回最终答案。
    return {"final_response": response}


# 定义决定是否继续的函数。
def should_continue(state: AgentState):
    messages = state["messages"]
    last_message = messages[-1]
    # 如果只有一个工具调用，并且是响应工具调用，我们就会回复用户。
    if (
        len(last_message.tool_calls) == 1
        and last_message.tool_calls[0]["name"] == "WeatherResponse"
    ):
        return "respond"
    # 否则我们将再次使用工具节点。
    else:
        return "continue"


# 定义一个新图。
workflow = StateGraph(AgentState)

# 定义我们将循环切换的两个节点。
workflow.add_node("agent", call_model)
workflow.add_node("respond", respond)
workflow.add_node("tools", ToolNode(tools))

# 将入口点设置为 `agent`。
# 这意味着这个节点是第一个被调用的节点。
workflow.set_entry_point("agent")

# 我们现在添加一个条件边。
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

from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_core.messages import HumanMessage


# 定义调用模型的函数。
def call_model(state: AgentState):
    response = model_with_tools.invoke(state["messages"])
    # 我们返回一个列表，因为这将被添加到现有列表中。
    return {"messages": [response]}


# 定义一个响应用户的函数。
def respond(state: AgentState):
    # 我们称这个具有结构化输出的模型，以便每次返回给用户相同的格式。
    # state['messages'][-2] is the last ToolMessage in the convo, which we convert to a HumanMessage for the model to use
    # 我们也可以传递整个聊天记录，但这样可以节省令牌，因为我们关心的只是工具的输出结构。
    response = model_with_structured_output.invoke(
        [HumanMessage(content=state["messages"][-2].content)]
    )
    # 我们返回最终答案。
    return {"final_response": response}


# 定义决定是否继续的函数。
def should_continue(state: AgentState):
    messages = state["messages"]
    last_message = messages[-1]
    # 如果没有函数调用，那么我们就会回应用户。
    if not last_message.tool_calls:
        return "respond"
    # 否则如果有，我们继续。
    else:
        return "continue"


# 定义一个新图。
workflow = StateGraph(AgentState)

# 定义我们将循环之间的两个节点。
workflow.add_node("agent", call_model)
workflow.add_node("respond", respond)
workflow.add_node("tools", ToolNode(tools))

# 将入口点设置为 `agent`。
# 这意味着这个节点是第一个被调用的。
workflow.set_entry_point("agent")

# 我们现在添加一个条件边。
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