1. 项目概述
在当今AI技术快速发展的背景下,任务型代理(AI Agent)已成为提升工作效率的重要工具。本文将详细介绍如何使用LangGraph框架构建一个能够自主调用多种工具的任务型代理系统。这个系统能够理解用户需求,自动判断并调用合适的工具(如天气查询、数学计算等),最终整合结果返回给用户。
与市面上大多数只讲理论的教程不同,本文提供的是一个完整可运行的实例,包含约200行核心代码。通过这个实例,开发者可以快速掌握LangGraph Agent开发的核心技术,并根据实际需求进行功能扩展。
2. 核心架构设计
2.1 系统工作流程
我们的Agent系统采用状态机模式设计,主要包含以下几个核心组件:
- Agent节点:负责调用大语言模型进行推理,判断是否需要调用工具
- Tools节点:执行具体的工具调用操作
- 状态管理器:维护对话上下文和工具调用结果
- 条件判断器:决定流程走向(继续调用工具或结束)
工作流程如下图所示(文字描述):
code复制用户提问 → [Agent节点] → LLM判断 → 需要工具? → [Tools节点] → 执行工具 → 返回结果
↑ ↓
└───────────────────────────────────┘
2.2 关键技术选型
选择LangGraph作为开发框架主要基于以下考虑:
- 显式流程控制:相比LangChain的Agent,LangGraph提供了更清晰的工作流定义方式
- 循环支持:可以轻松实现多次工具调用的场景
- 状态管理:内置的状态机机制让复杂交互场景的开发更简单
- 调试友好:每个节点的输入输出都清晰可见,便于问题排查
3. 环境准备与工具定义
3.1 开发环境配置
首先需要安装必要的Python包:
bash复制pip install langchain langgraph openai python-dotenv
建议使用Python 3.8+版本,并确保已设置好OpenAI API密钥。可以通过.env文件管理敏感信息:
python复制# .env文件内容
OPENAI_API_KEY=your_api_key_here
OPENAI_API_BASE=https://api.openai.com/v1
3.2 工具函数实现
工具是Agent能力的扩展,以下是两个基础工具的实现示例:
python复制def get_weather(city: str) -> str:
"""查询城市今日天气
Args:
city: 城市名称,如"北京"
Returns:
返回天气信息字符串
"""
# 实际项目中应接入真实天气API
weather_data = {
"北京": "28°C,晴朗",
"上海": "26°C,多云"
}
return weather_data.get(city, f"未找到{city}天气数据")
def calculate(expression: str) -> str:
"""计算数学表达式
Args:
expression: 数学表达式,如"3*5+2"
Returns:
计算结果字符串
"""
try:
return str(eval(expression))
except Exception as e:
return f"计算错误: {e}"
重要提示:工具函数的docstring和类型标注非常重要,它们会帮助LLM正确理解和使用工具。
4. Agent核心实现
4.1 状态定义
使用TypedDict明确定义Agent的状态结构:
python复制from typing import Annotated, Sequence, TypedDict
from langchain_core.messages import BaseMessage
import operator
class AgentState(TypedDict):
"""Agent运行时状态定义"""
messages: Annotated[Sequence[BaseMessage], operator.add]
这里使用operator.add确保消息是追加而非替换,这对维护完整的对话历史至关重要。
4.2 工作流构建
构建Agent的核心工作流程:
python复制def build_agent():
# 初始化LLM
llm = ChatOpenAI(
model="gpt-3.5-turbo",
temperature=0 # Agent任务需要确定性
).bind_tools(tools)
# 定义条件判断函数
def should_continue(state: AgentState):
last_message = state["messages"][-1]
return "continue" if getattr(last_message, "tool_calls", None) else END
# 构建工作流
workflow = StateGraph(AgentState)
# 添加节点
workflow.add_node("agent", lambda state: {"messages": [llm.invoke(state["messages"])]})
workflow.add_node("tools", ToolNode(tools))
# 设置流程关系
workflow.set_entry_point("agent")
workflow.add_conditional_edges("agent", should_continue, {"continue": "tools", END: END})
workflow.add_edge("tools", "agent")
return workflow.compile()
4.3 运行与交互
实现简单的命令行交互界面:
python复制def run_agent(app, query: str):
from langchain_core.messages import HumanMessage
print(f"\n用户: {query}")
for chunk in app.stream({"messages": [HumanMessage(content=query)]}, stream_mode="updates"):
for node, update in chunk.items():
if "messages" in update:
last_msg = update["messages"][-1]
if hasattr(last_msg, "tool_calls"):
for tc in last_msg.tool_calls:
print(f"调用工具: {tc['name']} 参数: {tc['args']}")
return chunk[-1][1]["messages"][-1].content
5. 常见问题与解决方案
5.1 工具调用异常排查
问题现象:工具被错误调用或参数格式不正确
解决方案:
- 确保工具函数有完整的docstring和参数类型标注
- 使用
bind_tools()方法正确绑定工具描述 - 检查LLM的温度参数(推荐temperature=0)
5.2 状态维护问题
问题现象:工具调用结果没有正确传递到下一步
解决方案:
- 确认状态定义使用了
Annotated[Sequence[BaseMessage], operator.add] - 检查是否遗漏了
import operator - 确保工具节点正确返回了结果消息
5.3 性能优化建议
- 批量处理:对于多个工具调用需求,考虑批量处理
- 缓存机制:对频繁查询的工具结果实施缓存
- 超时控制:为工具调用设置合理的超时时间
6. 扩展与进阶
6.1 添加新工具
扩展Agent功能只需三步:
- 实现新的工具函数
- 添加到tools列表
- 确保有清晰的docstring和类型标注
例如添加股票查询工具:
python复制def get_stock_price(symbol: str) -> str:
"""查询股票当前价格
Args:
symbol: 股票代码,如"AAPL"
Returns:
当前价格字符串
"""
# 实现实际的API调用
return f"{symbol} 当前价格: $150.2"
6.2 多Agent协作
LangGraph支持构建更复杂的多Agent系统:
python复制# 示例:构建两个协作Agent
workflow = StateGraph(AgentState)
workflow.add_node("research_agent", research_agent)
workflow.add_node("analysis_agent", analysis_agent)
workflow.add_edge("research_agent", "analysis_agent")
6.3 生产环境部署
将Agent部署为API服务的建议方案:
- 使用FastAPI构建REST接口
- 添加认证和限流中间件
- 实现异步处理长时间任务
- 添加日志和监控
7. 最佳实践总结
在实际项目中应用本方案时,建议:
-
工具设计原则:
- 每个工具应保持单一职责
- 输入输出尽量简单明确
- 包含充分的错误处理
-
流程优化技巧:
- 对频繁使用的工具设置优先级
- 考虑添加用户确认环节
- 实现结果缓存机制
-
调试建议:
- 使用
stream_mode="debug"查看详细流程 - 记录完整的对话历史
- 对关键节点添加日志
- 使用
通过本教程,开发者可以快速构建出功能强大的任务型Agent。这个基础框架可以根据实际需求进行各种扩展,比如添加更多工具、实现多Agent协作,或者集成到更大的系统中。