functools.partial 是什么

在 Python 中，partial 方法是 functools 模块中的一个功能，它用于创建一个新的函数，这个函数是基于原函数的部分参数已经固定的版本。这在需要重复调用同一函数，并且传递相同的某些参数时非常有用。

通过 partial，我们可以预先为函数的某些参数赋值，生成一个新的函数，这个新函数已经预先固定了部分参数，只需要再传递剩下的参数即可。

functools.partial 在 create_agent 函数中的应用

在你提供的代码中，partial 是用在 ChatPromptTemplate 的 from_messages 函数生成的提示模板上，它通过 partial 方法预先固定了部分参数。

prompt = prompt.partial(system_message=system_message)
prompt = prompt.partial(tool_names=", ".join([tool.name for tool in tools]))

这里的 partial 用于创建一个新的提示模板对象，并为 system_message 和 tool_names 这两个参数提供了值。这相当于对提示模板的“定制”，预先指定了这些参数的值。

partial 的具体作用：

1. 调用 prompt.partial(system_message=system_message)，预先为 system_message 参数赋值，生成一个新的提示模板，固定了系统消息的内容。
2. 调用 prompt.partial(tool_names=", ".join([tool.name for tool in tools]))，为 tool_names 参数赋值，将所有工具的名称合并成一个字符串，并固定在新的模板中。

通过这两步 partial 调用，prompt 对象中已经预先填入了 system_message 和 tool_names 这两个参数，简化了后续的调用过程。

---

functools.partial 在 research_node 定义时的应用

在这段代码中，functools.partial 用来创建一个新的函数，该函数已经预先绑定了部分参数。这种用法简化了后续调用函数时需要传递的参数，尤其是在重复使用某些固定参数时，partial 可以大大减少代码冗余。

research_node = functools.partial(agent_node, agent=research_agent, name="Researcher")

这里的 functools.partial 创建了一个新的函数 research_node，该函数基于原始的 agent_node 函数，且已经为 agent_node 的部分参数（agent 和 name）预先设置了值。新的 research_node 函数只需要接收剩余的参数就可以正常运行。

partial 的具体作用：

1. 原始函数 agent_node：

   def agent_node(state, agent, name):
       # 函数体...
   

   • agent_node 是一个接受 state, agent, 和 name 三个参数的函数。

2. 使用 functools.partial：

   research_node = functools.partial(agent_node, agent=research_agent, name="Researcher")
   

   • 通过 functools.partial，我们创建了一个新的函数 research_node，它仍然是 agent_node，但 agent 参数和 name 参数已经被预先固定：
     • agent=research_agent
     • name="Researcher"
   • 也就是说，调用 research_node 时，只需要传递 state 参数，因为 agent 和 name 已经有默认值了。

举个例子

假设有一个函数 agent_node，你经常需要调用它并传递相同的 agent 和 name，那么每次调用时重复写这些参数会很冗余。使用 partial 可以避免这种重复。

# 原始函数定义
def agent_node(state, agent, name):
    print(f"State: {state}, Agent: {agent}, Name: {name}")

# 预先设置 agent 和 name 参数
research_node = functools.partial(agent_node, agent="research_agent_value", name="Researcher")

# 调用时只需要传递剩下的参数
research_node(state="current_state")
# 输出: State: current_state, Agent: research_agent_value, Name: Researcher

functools.partial 的优势

1. 减少重复代码：在你需要多次调用同一个函数并且某些参数不变时，partial 可以避免每次都传递相同的参数。

2. 简化函数调用：在需要频繁使用相同参数时，partial 提供了更简洁的写法，使代码更易于维护。

总结

在这段代码中，functools.partial 的用法预先为 agent_node 函数的部分参数（agent 和 name）赋值，创建了一个新函数 research_node。调用 research_node 时，只需要传递剩下的参数（state），从而简化了函数调用的流程。

---

什么是 ToolNode？

ToolNode 是 LangChain 的一个预构建节点，它能够从图状态（graph state）中提取消息并调用指定的工具，最后将工具调用的结果反馈回图的状态中。ToolNode 非常适合与 LangGraph 中的 ReAct agent 协同工作，但也可以与任何 StateGraph 配合使用，只要状态中有 messages 键和合适的消息处理方式。

ToolNode 的特点

1. 工具调用：ToolNode 可以根据状态中的消息自动调用指定的工具，并返回工具的执行结果。
2. 兼容性：可以与任意支持工具调用的 LangChain 模型配合使用。
3. 并行工具调用：支持同时调用多个工具，并处理工具返回的多个结果。
4. 错误处理：ToolNode 默认启用了错误处理，可以处理工具在执行过程中的异常情况。

ToolNode 的使用步骤

1. 安装和环境设置

首先，安装所需的包并设置 API 密钥：

%%capture --no-stderr
%pip install --quiet -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")

2. 定义工具

使用 @tool 装饰器定义可以被 ToolNode 调用的工具。下面的例子定义了两个工具：

• get_weather：获取某个地点的天气。
• get_coolest_cities：获取最酷的城市列表。

from langchain_core.messages import AIMessage
from langchain_core.tools import tool
from langgraph.prebuilt import ToolNode

@tool
def get_weather(location: str):
    """获取当前的天气。"""
    if location.lower() in ["sf", "san francisco"]:
        return "It's 60 degrees and foggy."
    else:
        return "It's 90 degrees and sunny."

@tool
def get_coolest_cities():
    """获取最酷的城市列表"""
    return "nyc, sf"

tools = [get_weather, get_coolest_cities]  # 将定义的工具放入列表中
tool_node = ToolNode(tools)  # 使用工具列表初始化 ToolNode

3. 手动调用 ToolNode

ToolNode 通过图状态操作消息列表，它期望列表中的最后一条消息是一个带有 tool_calls 参数的 AIMessage。以下是手动调用 ToolNode 的示例：

message_with_single_tool_call = AIMessage(
    content="",
    tool_calls=[
        {
            "name": "get_weather",
            "args": {"location": "sf"},
            "id": "tool_call_id",
            "type": "tool_call",
        }
    ],
)

tool_node.invoke({"messages": [message_with_single_tool_call]})
# 返回: {'messages': [ToolMessage(content="It's 60 degrees and foggy.", name='get_weather', tool_call_id='tool_call_id')]}

4. 并行调用多个工具

ToolNode 也支持并行调用多个工具，只需在 AIMessage 的 tool_calls 参数中传入多个工具调用：

message_with_multiple_tool_calls = AIMessage(
    content="",
    tool_calls=[
        {
            "name": "get_coolest_cities",
            "args": {},
            "id": "tool_call_id_1",
            "type": "tool_call",
        },
        {
            "name": "get_weather",
            "args": {"location": "sf"},
            "id": "tool_call_id_2",
            "type": "tool_call",
        },
    ],
)

tool_node.invoke({"messages": [message_with_multiple_tool_calls]})
# 返回:
# {'messages': [ToolMessage(content='nyc, sf', name='get_coolest_cities', tool_call_id='tool_call_id_1'),
#   ToolMessage(content="It's 60 degrees and foggy.", name='get_weather', tool_call_id='tool_call_id_2')]}

与对话模型结合使用

在使用像 Anthropic 这样的对话模型时，模型可以自动生成带有 tool_calls 的 AIMessage，这样我们可以直接将模型生成的消息传给 ToolNode 来执行工具调用：

from langchain_anthropic import ChatAnthropic
from langgraph.prebuilt import ToolNode

model_with_tools = ChatAnthropic(
    model="claude-3-haiku-20240307", temperature=0
).bind_tools(tools)

tool_node.invoke({"messages": [model_with_tools.invoke("what's the weather in sf?")]})
# 返回: {'messages': [ToolMessage(content="It's 60 degrees and foggy.", name='get_weather', tool_call_id='toolu_01LFvAVT3xJMeZS6kbWwBGZK')]}

ToolNode 与 ReAct Agent 结合

ReAct Agent 是 LangGraph 中的一种智能体，它会反复调用工具，直到收集到足够的信息来解决问题。以下是 ReAct Agent 的基本工作流，它通过工具节点来完成工具调用：

from typing import Literal
from langgraph.graph import StateGraph, MessagesState

def should_continue(state: MessagesState) -> Literal["tools", "__end__"]:
    messages = state["messages"]
    last_message = messages[-1]
    if last_message.tool_calls:
        return "tools"
    return "__end__"

def call_model(state: MessagesState):
    messages = state["messages"]
    response = model_with_tools.invoke(messages)
    return {"messages": [response]}

# 创建状态图
workflow = StateGraph(MessagesState)

# 定义两个节点：一个用于调用模型，一个用于调用工具
workflow.add_node("agent", call_model)
workflow.add_node("tools", tool_node)

workflow.add_edge("__start__", "agent")  # 从 agent 节点开始
workflow.add_conditional_edges("agent", should_continue)  # 根据条件判断是否继续调用工具
workflow.add_edge("tools", "agent")  # 工具调用完成后，返回 agent 节点

app = workflow.compile()  # 编译状态图

例子：调用单个工具

当用户输入 "what's the weather in sf?" 时，智能体将调用 get_weather 工具并返回天气信息：

for chunk in app.stream(
    {"messages": [("human", "what's the weather in sf?")]}, stream_mode="values"
):
    chunk["messages"][-1].pretty_print()

例子：连续调用多个工具

当用户输入 "what's the weather in the coolest cities?" 时，智能体将依次调用 get_coolest_cities 和 get_weather 工具，返回所有城市的天气信息：

for chunk in app.stream(
    {"messages": [("human", "what's the weather in the coolest cities?")]},
    stream_mode="values",
):
    chunk["messages"][-1].pretty_print()

错误处理

ToolNode 默认启用了错误处理，可以处理工具执行中的异常情况。如果想禁用错误处理，可以设置 handle_tool_errors=False。

总结

ToolNode 是一个非常强大的组件，它能够自动调用工具并将结果反馈回工作流。它可以处理单个或多个工具调用，并与 LangChain 模型紧密结合，使得在复杂的多步骤任务中能够更高效地调用外部 API 或工具。