AI 代理(AI Agents)与传统聊天机器人最大的不同在于:它们不仅能交流,还能在后台自主调用工具,并根据环境反馈决定接下来的行动。
本指南将深入探讨如何使用 LangGraph 构建有状态的、生产级别的 AI 代理,并解析其底层的运行逻辑。
核心概念
在使用 LangGraph 开发代理时,你需要掌握三个核心支柱:
- 工具 (Tools):代理可以执行的 Python 函数。
- 状态 (State):代理在运行过程中携带的上下文信息(如对话历史)。
- 图 (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()
代理的工作流演示
当用户提问:“我本月在食品杂货上花了多少钱?”时,程序的执行路径如下:
- [Agent Node]:LLM 接收问题,查看
search_transactions工具,决定调用它并传入参数category='groceries'。 - [Should Continue]:检测到
tool_calls,路由至tools。 - [Tool Node]:执行数据库查询,获得数据及摘要,将其作为
ToolMessage存入状态。 - [Agent Node]:LLM 重新审视上下文(问题 + 工具调用记录 + 工具返回结果)。
- [Agent Node]:LLM 意识到已经拥有足够信息,生成最终文本回复。
- [Should Continue]:未检测到新的
tool_calls,路由至END。
总结
使用 LangGraph 开发代理的精髓在于:明确定义规则,而非寄希望于黑盒。
- 工具 提供了能力。
- 状态 提供了记忆。
- 图 提供了执行的确定性。
通过这种结构化方案,你可以构建出极其复杂且可靠的 AI 系统,应对金融、法律或技术支持等严肃场景。