利用 LangGraph 和 LLaMA 3 构建多 Agent 助手：新手入门指南

在构建复杂的 AI 应用时，我们常常需要将任务分解为多个独立的、可控的模块。传统的单体模型往往难以胜任这种精细化的需求。本文将深入探讨如何使用 LangGraph 这一强大的框架，结合 Groq 提供的 LLaMA 3 大模型，以及 LangChain 工具，构建一个高效且可扩展的 多 Agent 助手。我们将从 LangGraph 的核心概念入手，逐步解析其组件，并通过代码示例，展示如何一步步构建一个能够处理数学计算、网络搜索和礼貌道别的 多 Agent 助手。

LangGraph：构建对话蓝图的核心思想

LangGraph 的核心思想是将复杂的对话流程拆解为一系列独立的节点 (Nodes)，每个节点负责执行特定的任务。我们可以将其比作一个对话蓝图，其中每个转折点（节点）都有明确的目的，而节点之间的流动方向则由用户的输入或模型的预测来决定。不同于编写单一冗长的逻辑链，LangGraph 允许我们将系统分解为多个独立的模块，并负责管理状态和决策流程。这种模块化的设计使得开发、调试和扩展 AI 助手变得更加容易。例如，一个客户服务机器人可以被分解为以下几个节点：

问候节点： 负责与用户建立初步联系，询问其需求。
意图识别节点： 利用 LLaMA 3 等模型识别用户的意图，例如“查询订单状态”、“修改收货地址”等。
订单查询节点： 调用数据库 API 查询用户的订单信息。
地址修改节点： 调用地址修改 API 修改用户的收货地址。
结束节点： 结束对话，感谢用户的支持。

通过 LangGraph，我们可以轻松地将这些节点连接起来，构建一个完整的客户服务流程。

关键概念解析：状态、节点、边和监督器

要理解 LangGraph 的工作原理，需要掌握以下几个关键概念：

状态 (State): 可以将状态想象成一个背包，它存储着对话过程中产生的所有信息，例如用户的消息、助手的回复、决策标志等。每个 LangGraph 中的节点都会接收这个背包，进行处理（例如添加新消息），然后将更新后的状态传递给下一个节点。状态的管理对于构建复杂的对话流程至关重要，它确保了每个节点都能访问到必要的信息，从而做出正确的决策。例如，在一个电商助手中，状态可以包含用户的购物车信息、浏览历史、收货地址等。
节点 (Node): 一个节点就是一个简单的函数。它接收状态作为输入，执行特定的任务，然后返回更新后的状态。节点是 LangGraph 中最基本的单元，每个节点都应该专注于完成一个明确的任务，例如数学计算、网络搜索或数据库查询。例如，文章中提供的 calculator 函数就是一个节点，它接收包含数学表达式的状态，计算结果并返回包含结果的状态。
边 (Edge): 边连接两个节点，定义了节点之间的流动方向。边可以是简单的，例如总是从节点 A 流向节点 B；也可以是条件性的，根据状态的值来决定下一个节点。条件边允许我们构建智能的路由逻辑，让系统根据用户的输入或模型的预测，动态地选择下一个执行的节点。例如，在一个医疗助手中，可以根据用户的症状描述，选择不同的诊断节点。
监督器 (Supervisor) / 路由器 (Router): 监督器是 AI 助手的“大脑”。它分析用户的输入，并对其进行分类，例如：这是一个数学问题？一个一般性查询？还是仅仅是用户在说“谢谢”？基于分类结果，监督器选择下一个执行的节点。在文章的例子中，监督器利用 LLaMA 3 模型对用户的输入进行分类，并根据分类结果选择 math_node、search_node 或 farewell 节点。监督器的准确性直接影响到 AI 助手的性能，因此需要精心设计其提示语和分类逻辑。
入口 (Entry) 和出口 (Exit): 每个图都需要一个起始点（称为入口节点）。并且每个流程在某个时候结束。 LangGraph 提供了一个 END 标记来在路径完成后干净地完成。

优势：模块化、可扩展性、透明性和可控性

LangGraph 的设计带来了诸多优势：

模块化 (Modularity): 每个函数（节点）都专注于完成一个简单的任务，降低了代码的复杂度，提高了可维护性。
可扩展性 (Scalability): 可以轻松地添加新的查询类型或行为，而无需修改现有的代码。例如，我们可以轻松地添加一个新的节点来处理用户上传的图片，或者添加一个新的节点来提供个性化的推荐。
透明性 (Transparency): 可以可视化和调试流程，更好地理解系统的行为。 LangGraph 提供了可视化工具，可以清晰地展示节点之间的连接关系和状态的流动方向，帮助开发者快速定位问题。
可控性 (Control): 可以显式地定义逻辑在何处发生，而不是依赖于一个模型来神奇地完成所有事情。这种控制力使得开发者可以更好地掌握系统的行为，并进行精细化的调整。

代码实践：构建一个多 Agent 助手

接下来，我们将通过代码示例，展示如何使用 LangGraph、LLaMA 3 和 LangChain 工具构建一个 多 Agent 助手。

步骤 1：导入必要的库

import os
import operator
import re
from typing import TypedDict, Literal, Annotated
import langgraph
from langgraph.graph import StateGraph, END
from langchain_groq import ChatGroq
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain.tools import tool
from langchain_community.tools.tavily_search import TavilySearchResults
from pydantic import BaseModel

这段代码导入了构建 多 Agent 助手 所需的各种库，包括：

os: 用于访问环境变量，例如 API 密钥。
re: 用于使用正则表达式处理数学表达式。
typing: 用于定义 LangGraph 状态的严格模式及其合并行为。
langgraph: 提供定义图（节点 + 流程）的工具。
langchain_groq: 允许调用 Groq 托管的模型，例如 LLaMA 3。
langchain_core.messages: 用于模拟对话过程中交换的消息。
langchain.tools.tool: 用于将任何自定义函数注册为 LangChain 工具。
langchain_community.tools.tavily_search: 用于使用 Tavily API 获取网络搜索结果的内置 LangChain 工具。
pydantic: 用于定义具有模型输出验证的模式（例如来自路由器的结构化 JSON）。

步骤 2：定义共享状态

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    step: str

这段代码定义了在节点之间传递的共享状态。messages 存储消息列表（例如聊天记录）。Annotated[list, operator.add] 告诉 LangGraph 如何在需要时组合消息（通过附加）。step 存储一个字符串，例如 “math_node” 或 “search_node”，告诉图接下来应该执行哪个节点。

步骤 3：加载 LLaMA 3 模型

groq_model = ChatGroq(
    model="llama3-70b-8192",
    temperature=0.2,
    api_key=os.getenv("GROQ_API_KEY")
)

这段代码初始化一个 Groq 客户端，用于与 LLaMA 3 模型进行通信。temperature 参数设置为较低的值，以确保确定性行为（这对于路由非常有用）。它使用存储在环境变量中的 API 密钥。

步骤 4：定义工具

@tool
def calculator(expression: str) -> str:
    """Calculates a simple math expression."""
    try:
        expression = expression.strip()
        expression = expression.replace(' ', '')
        numbers = re.findall(r'\d+', expression)
        numbers = [int(number) for number in numbers]
        operator_match = re.search(r'[+\-*\/]', expression)
        operator = operator_match.group(0) if operator_match else None
        if not operator:
            return "Error: No operator found in the expression."
        if operator == '+':
            result = numbers[0] + numbers[1]
        elif operator == '-':
            result = numbers[0] - numbers[1]
        elif operator == '*':
            result = numbers[0] * numbers[1]
        elif operator == '/':
            if numbers[1] == 0:
                return "Error: Division by zero."
            result = numbers[0] / numbers[1]
        return str(result)
    except Exception as e:
        return f"Error: {str(e)}"


try:
    tavily_search = TavilySearchResults(max_results=2)
    TAVILY_AVAILABLE = True
except:
    TAVILY_AVAILABLE = False

@tool
def fallback_search(query: str) -> str:
    """A fallback search tool when Tavily API key is not configured."""
    return f"I would search for '{query}' but Tavily API key is not configured."

这段代码定义了两个工具：calculator 用于计算数学表达式，tavily_search 用于进行网络搜索。如果 Tavily API 密钥未配置，则使用 fallback_search 作为替代方案。

步骤 5：定义路由器模式和提示语

ROUTER_PROMPT = """Classify the user's input and pick one category:
- If the user asks a math question, select "math_node".
- If the user asks a real-world question, select "search_node".
- If the user says goodbye, select "farewell".
"""

class Router(BaseModel):
    next: Literal["search_node", "math_node", "farewell"]
    reasoning: str

这段代码定义了路由器模式和提示语。ROUTER_PROMPT 用于指导模型对用户的输入进行分类。Router 类定义了模型返回的 JSON 结构，包括 next (下一个执行的节点) 和 reasoning (选择该节点的理由)。

步骤 6：定义节点

def supervisor_node(state: AgentState):
    messages = state["messages"]
    response = groq_model.with_structured_output(Router).invoke({"messages": [
        SystemMessage(content=ROUTER_PROMPT),
        *messages
    ]})
    return {"step": response.next}

def math_node(state: AgentState):
    messages = state["messages"]
    query = messages[-1].content
    result = calculator.invoke({"expression": query})
    response = f"The answer is: {result}"
    return {"messages": [AIMessage(content=response)], "step": "farewell"}

def search_node(state: AgentState):
    messages = state["messages"]
    query = messages[-1].content
    if not TAVILY_AVAILABLE:
        results = fallback_search.invoke({"query": query})
    else:
        results = tavily_search.invoke({"query": query})
    response = f"Here are the search results: {results}"
    return {"messages": [AIMessage(content=response)], "step": "farewell"}

def farewell_node(state: AgentState):
    return {"messages": [AIMessage(content="Goodbye!")]}

这段代码定义了四个节点：supervisor_node (监督器节点)，math_node (数学计算节点)，search_node (网络搜索节点) 和 farewell_node (道别节点)。每个节点都接收状态作为输入，执行特定的任务，然后返回更新后的状态。

步骤 7：设置 LangGraph 并定义流程

def route_supervisor(state: AgentState):
    return state["step"]

graph = StateGraph(AgentState)
graph.set_entry_point("supervisor")
graph.add_node("supervisor", supervisor_node)
graph.add_node("math_node", math_node)
graph.add_node("search_node", search_node)
graph.add_node("farewell", farewell_node)

graph.add_conditional_edges(
    "supervisor", route_supervisor,
    {
        "math_node": "math_node",
        "search_node": "search_node",
        "farewell": "farewell"
    }
)

graph.add_edge("math_node", "farewell")
graph.add_edge("search_node", "farewell")
graph.add_edge("farewell", END)

app = graph.compile()

这段代码设置 LangGraph 并定义流程。route_supervisor 函数根据状态中的 step 字段选择下一个执行的节点。graph 对象是 LangGraph 的核心，它将所有节点和边连接起来。app 对象是编译后的 LangGraph 应用，可以被调用来执行对话流程。

步骤 8：运行助手

initial_state = {
    "messages": [HumanMessage(content="What's 15*9?")],
    "step": "supervisor"
}

result = app.invoke(initial_state)

这段代码定义了初始状态，并调用 app.invoke() 方法来运行助手。

步骤 9：打印输出

print("\n--- Conversation ---")
for msg in result["messages"]:
    print(f"{msg.__class__.__name__}: {msg.content}")

这段代码打印对话历史记录，以便理解助手的工作过程。

总结：LangGraph 的未来

通过本文的介绍，我们了解了如何使用 LangGraph、LLaMA 3 和 LangChain 工具构建一个 多 Agent 助手。LangGraph 提供了一种模块化、可扩展和可控的方式来构建复杂的 AI 应用。随着大模型技术的不断发展，LangGraph 将在构建更加智能、更加个性化的 AI 助手方面发挥越来越重要的作用。我们期待 LangGraph 在未来能够支持更多的特性，例如：

更强大的可视化工具： 提供更丰富的可视化功能，帮助开发者更好地理解和调试复杂的对话流程。
更灵活的路由机制： 支持更复杂的路由规则，例如基于上下文的路由和基于用户角色的路由。
更丰富的节点类型： 提供更多预定义的节点类型，例如数据库查询节点、API 调用节点和自然语言生成节点。

通过不断地完善和发展，LangGraph 有望成为构建下一代 AI 应用的标准框架。

利用 LangGraph 和 LLaMA 3 构建多 Agent 助手：新手入门指南