Tools & 多 Agent 架构

7.1 工具定义基础

LangGraph 使用 @tool 装饰器定义工具，自动生成 Schema 并绑定到 LLM。

7.1.1 基础工具定义

python
from langchain_core.tools import tool

@tool
def search_web(query: str) -> str:
    """搜索互联网获取最新信息

    Args:
        query: 搜索关键词，越具体越好
    """
    # 实际项目中调用搜索 API
    return f"搜索结果：关于 {query} 的最新信息..."

@tool
def calculate(expression: str) -> str:
    """执行数学计算

    Args:
        expression: 数学表达式，如 "2+3*5"
    """
    result = eval(expression)
    return str(result)

@tool
def get_weather(city: str) -> str:
    """查询城市天气

    Args:
        city: 城市名称，如"北京"、"上海"
    """
    return f"{city}今天晴，25°C"

7.1.2 工具节点工作流

7.1.3 绑定工具到 LLM

python
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

llm = ChatOpenAI(model="gpt-4o")
tools = [search_web, calculate, get_weather]

# 绑定工具：LLM 会自动判断何时调用工具
llm_with_tools = llm.bind_tools(tools)

# 调用示例
messages = [HumanMessage(content="北京今天天气怎么样？")]
response = llm_with_tools.invoke(messages)

# LLM 返回的是 AIMessage，包含 tool_calls
print(response.tool_calls)
# [{'name': 'get_weather', 'args': {'city': '北京'}, 'id': 'xxx'}]

7.2 工具节点实现

有了 tool_calls，需要工具节点来实际执行并返回结果。

7.2.1 标准工具节点

python
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]

def tool_node(state: AgentState) -> dict:
    """工具节点：执行 LLM 请求的工具调用"""
    last_message = state["messages"][-1]

    # 没有工具调用，直接返回
    if not hasattr(last_message, "tool_calls") or not last_message.tool_calls:
        return {"messages": []}

    results = []
    for tool_call in last_message.tool_calls:
        tool_name = tool_call["name"]
        tool_args = tool_call["args"]

        # 根据工具名找到对应函数并调用
        if tool_name == "search_web":
            result = search_web.invoke(tool_args)
        elif tool_name == "calculate":
            result = calculate.invoke(tool_args)
        elif tool_name == "get_weather":
            result = get_weather.invoke(tool_args)
        else:
            result = f"未知工具: {tool_name}"

        results.append(
            ToolMessage(content=result, tool_call_id=tool_call["id"])
        )

    return {"messages": results}

7.2.2 RAG 工具实战

python
@tool
def rag_search(query: str) -> str:
    """从知识库检索相关内容

    Args:
        query: 用户问题
    Returns:
        最相关的文档片段
    """
    docs = vectorstore.similarity_search(query, k=3)
    return "\n\n".join([doc.page_content for doc in docs])

# 完整的 Agent 图
graph = StateGraph(AgentState)
graph.add_node("llm", llm_node)
graph.add_node("tools", tool_node)
graph.add_edge(START, "llm")
graph.add_conditional_edges("llm", route_to_tools, {
    "tools": "tools",
    "__end__": END
})
graph.add_edge("tools", "llm")  # 工具结果返回 LLM
app = graph.compile()

7.3 多 Agent 架构

当单 Agent 难以完成复杂任务时，多 Agent 协作是解决方案。每个 Agent 专注特定领域，通过消息传递协作。

7.3.1 Supervisor 调度模式

一个中央 Supervisor 根据任务类型分配给专业 Agent。

python
from typing import Literal

class MultiAgentState(TypedDict):
    task: str
    current_agent: Literal["researcher", "coder", "reviewer", "__end__"] | None
    research_result: str | None
    code_result: str | None
    review_result: str | None

def supervisor_node(state: MultiAgentState) -> dict:
    """Supervisor 决定下一个处理者"""
    task = state["task"]

    if not state.get("research_result"):
        return {"current_agent": "researcher"}
    elif not state.get("code_result"):
        return {"current_agent": "coder"}
    elif not state.get("review_result"):
        return {"current_agent": "reviewer"}
    else:
        return {"current_agent": "__end__"}

def route_to_agent(state: MultiAgentState) -> str:
    return state["current_agent"]

def researcher_node(state: MultiAgentState) -> dict:
    result = do_research(state["task"])
    return {"research_result": result}

def coder_node(state: MultiAgentState) -> dict:
    code = write_code(state["task"], state.get("research_result"))
    return {"code_result": code}

def reviewer_node(state: MultiAgentState) -> dict:
    result = review_code(state.get("code_result", ""))
    return {"review_result": result}

# 构建图
graph = StateGraph(MultiAgentState)
graph.add_node("supervisor", supervisor_node)
graph.add_node("researcher", researcher_node)
graph.add_node("coder", coder_node)
graph.add_node("reviewer", reviewer_node)
graph.add_edge(START, "supervisor")
graph.add_conditional_edges("supervisor", route_to_agent, {
    "researcher": "researcher",
    "coder": "coder",
    "reviewer": "reviewer",
    "__end__": END
})
graph.add_edge("researcher", "supervisor")
graph.add_edge("coder", "supervisor")
graph.add_edge("reviewer", "supervisor")

7.3.2 并行执行模式

使用 Send API 同时调用多个 Agent，再聚合结果。

python
from langgraph.constants import Send

def parallel_research(state: MainState) -> list:
    """并行执行多个研究任务"""
    topic = state["topic"]
    return [
        Send("web_search", {"query": f"{topic} 最新动态"}),
        Send("academic_search", {"query": f"{topic} 学术论文"}),
        Send("social_search", {"query": f"{topic} 社区讨论"}),
    ]

graph.add_node("parallel_search", parallel_research)
graph.add_node("aggregate", aggregate_results)
graph.add_edge(START, "parallel_search")
graph.add_edge("parallel_search", "aggregate")
graph.add_edge("aggregate", END)

7.4 面试要点

7.5 本章小结

• Tools：@tool 装饰器定义工具，bind_tools 绑定到 LLM
• 工具节点：解析 tool_calls，执行对应函数，返回 ToolMessage
• 多 Agent：Supervisor 调度、Agent 间交接、并行执行（Send）
• RAG 集成：将向量检索包装为工具，让 LLM 动态检索知识
• 面试重点：Tool Calling 流程、多 Agent 架构对比、循环控制策略