使用 LangGraph 开发 AI 代理:实用指南

AI 代理(AI Agents)与传统聊天机器人最大的不同在于:它们不仅能交流,还能在后台自主调用工具,并根据环境反馈决定接下来的行动。

本指南将深入探讨如何使用 LangGraph 构建有状态的、生产级别的 AI 代理,并解析其底层的运行逻辑。


核心概念

在使用 LangGraph 开发代理时,你需要掌握三个核心支柱:

  1. 工具 (Tools):代理可以执行的 Python 函数。
  2. 状态 (State):代理在运行过程中携带的上下文信息(如对话历史)。
  3. 图 (Graph):代理的“大脑骨架”,定义了决策逻辑和操作顺序。

1. 定义工具 (Tools)

工具是代理与外部世界交互的唯一方式。LLM 通过函数的名称文档字符串(Docstrings)和参数模式来判断何时调用该工具。

优秀工具设计的原则

  • 详尽的文档字符串:这是 LLM 的“说明书”,描述越清晰,调用越准确。
  • 依赖隐藏:使用工厂函数模式注入数据库连接或 API Key,只给 LLM 暴露必要的参数。
from langchain_core.tools import tool
from sqlalchemy import select

def create_search_transactions_tool(search_space_id: int, db_session):
    @tool
    async def search_transactions(
        keywords: str | None = None,
        category: str | None = None
    ) -> dict:
        """根据商家或类别搜索财务交易。
        
        适用场景:
        - 查询在特定商家的消费(如:“我在星巴克花了多少钱?”)
        - 查询特定类别的支出(如:“本月买菜花了多少?”)
        """
        # 数据库查询逻辑...
        query = select(Document.document_metadata).where(
            Document.search_space_id == search_space_id
        )
        # 模拟返回结构
        all_transactions = [...] 
        total = sum(abs(t["amount"]) for t in all_transactions)
        
        return {
            "transactions": all_transactions[:20], 
            "total_amount": total,
            "summary": f"找到了 {len(all_transactions)} 笔交易,总计 ${total:,.2f}"
        }
    
    return search_transactions


2. 代理状态 (Agent State)

状态是图中流动的“真相来源”。它不仅存储对话历史,还可以记录重试次数、用户 ID 或当前步骤。

from typing import Annotated, Sequence, TypedDict
from langchain_core.messages import BaseMessage

class AgentState(TypedDict):
    # 使用 Annotated 告诉 Graph 如何处理新消息(例如:追加到列表末尾)
    messages: Annotated[Sequence[BaseMessage], "对话历史记录"]
    user_id: str
    retry_count: int


3. 构建代理图 (The Graph)

图将代理的逻辑流程化。主要包含三种元素:

  • 节点 (Nodes):具体的操作,如 call_agent(询问 LLM)或 tool_node(运行工具)。
  • 边缘 (Edges):连接节点的线。
  • 条件边缘 (Conditional Edges):根据逻辑判断下一步去往哪个节点。

步骤 1:代理决策节点

async def call_agent(state: AgentState):
    messages = state["messages"]
    # 将工具绑定到 LLM
    llm_with_tools = llm.bind_tools(tools)
    # 调用 LLM 做出决策
    response = await llm_with_tools.ainvoke(messages)
    # 更新状态:将 LLM 的回答追加到消息列表
    return {"messages": [response]}

步骤 2:控制流逻辑

from langgraph.graph import END

def should_continue(state: AgentState):
    last_message = state["messages"][-1]
    # 如果 LLM 的最新消息包含 tool_calls,则进入工具节点
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"
    # 否则,任务结束
    return END

步骤 3:组装图结构

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

workflow = StateGraph(AgentState)

# 1. 添加节点
workflow.add_node("agent", call_agent)
workflow.add_node("tools", ToolNode(tools))

# 2. 设置入口
workflow.set_entry_point("agent")

# 3. 设置路径逻辑
workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "tools": "tools", # 如果逻辑返回 "tools",前往工具节点
        END: END          # 如果逻辑返回 END,结束流程
    }
)

# 4. 工具执行完后,必须回到代理节点进行结果分析
workflow.add_edge("tools", "agent")

# 5. 编译
app = workflow.compile()


代理的工作流演示

当用户提问:“我本月在食品杂货上花了多少钱?”时,程序的执行路径如下:

  1. [Agent Node]:LLM 接收问题,查看 search_transactions 工具,决定调用它并传入参数 category='groceries'
  2. [Should Continue]:检测到 tool_calls,路由至 tools
  3. [Tool Node]:执行数据库查询,获得数据及摘要,将其作为 ToolMessage 存入状态。
  4. [Agent Node]:LLM 重新审视上下文(问题 + 工具调用记录 + 工具返回结果)。
  5. [Agent Node]:LLM 意识到已经拥有足够信息,生成最终文本回复。
  6. [Should Continue]:未检测到新的 tool_calls,路由至 END

总结

使用 LangGraph 开发代理的精髓在于:明确定义规则,而非寄希望于黑盒。

  • 工具 提供了能力。
  • 状态 提供了记忆。
  • 提供了执行的确定性。

通过这种结构化方案,你可以构建出极其复杂且可靠的 AI 系统,应对金融、法律或技术支持等严肃场景。