1. 基于LangGraph的计算器代理开发实战
在AI应用开发领域,构建能够理解并执行特定任务的智能代理已成为提升工作效率的热门方向。今天我要分享的是如何使用LangGraph框架开发一个能处理基础算术运算的计算器代理。这个项目虽然看似简单,但完整展示了从工具定义、模型绑定到工作流构建的全过程,是理解AI代理开发范式的绝佳入门案例。
我选择LangChain生态的LangGraph模块来实现,主要基于三点考虑:首先,LangChain提供了丰富的工具链和模型集成能力;其次,Graph API的声明式编程方式让复杂逻辑可视化;最后,Python生态的易用性可以快速验证想法。这个代理最终能处理加减乘除运算,并能通过自然语言交互,比如当用户询问"3加4等于多少"时,它能准确理解意图并返回计算结果。
需要模型API调用? 免费领10W Token,多模型网关一键接入 Claude、DeepSeek 等主流模型。
2. 环境准备与工具定义
2.1 开发环境配置
建议使用Python 3.11.x版本以获得最佳兼容性,这是我经过多个版本测试后的经验之选。依赖管理推荐使用poetry或pipenv,以下是必须安装的核心包:
bash复制pip install langchain langgraph litellm
注意:litellm作为模型抽象层,可以无缝切换不同厂商的API。本文示例使用阿里云百炼平台的Qwen模型,但你也可以替换为OpenAI或本地部署的模型。
2.2 算术工具函数实现
工具函数是代理的能力基础,我们需要用@tool装饰器明确定义每个运算的输入输出:
python复制from langchain.tools import tool
@tool
def multiply(a: int, b: int) -> int:
"""Multiply two integers `a` and `b`.
Args:
a: First integer
b: Second integer
"""
return a * b
@tool
def add(a: int, b: int) -> int:
"""Add two integers `a` and `b`."""
return a + b
@tool
def divide(a: int, b: int) -> float:
"""Divide `a` by `b`. Returns float."""
return a / b
几个关键细节:
- 类型注解必须明确(如
int/float),这会影响LLM的参数解析 - 文档字符串要简洁准确,它会被用作工具的说明
- 除法建议返回float以避免精度丢失
3. 模型初始化与工具绑定
3.1 大语言模型配置
通过LiteLLM统一接口初始化模型,这里以阿里云Qwen为例:
python复制from langchain.chat_models import init_chat_model
model = init_chat_model(
model="dashscope/qwen-math-turbo",
model_provider="litellm",
temperature=0, # 确定性输出
model_kwargs={
"api_key": "your_api_key",
}
)
实操技巧:设置temperature=0对数学计算至关重要,更高的值可能导致结果不稳定。
3.2 工具能力注入
将工具列表绑定到模型,创建增强版LLM实例:
python复制tools = [add, multiply, divide]
tools_by_name = {tool.name: tool for tool in tools}
model_with_tools = model.bind_tools(tools)
绑定后模型会:
- 自动识别工具的使用场景
- 在适当时候生成工具调用请求
- 理解工具返回的结果
4. 工作流构建与状态管理
4.1 状态类型定义
使用TypedDict明确工作流的状态结构:
python复制from typing_extensions import TypedDict, Annotated
import operator
class MessagesState(TypedDict):
messages: Annotated[list[AnyMessage], operator.add]
llm_calls: int
messages:累积所有消息(用户输入、AI回复、工具结果)llm_calls:记录LLM调用次数(用于监控和调试)
4.2 核心节点实现
4.2.1 LLM决策节点
python复制from langchain.messages import SystemMessage
def llm_call(state: dict):
"""决定是否调用工具"""
return {
"messages": [
model_with_tools.invoke([
SystemMessage(content="你是一个数学助手,专注算术运算")
] + state["messages"])
],
"llm_calls": state.get('llm_calls', 0) + 1
}
系统消息(SystemMessage)用于设定AI的角色和行为准则,这是控制输出质量的关键。
4.2.2 工具执行节点
python复制from langchain.messages import ToolMessage
def tool_node(state: dict):
"""执行具体的工具调用"""
results = []
for tool_call in state["messages"][-1].tool_calls:
tool = tools_by_name[tool_call["name"]]
observation = tool.invoke(tool_call["args"])
results.append(
ToolMessage(content=str(observation),
tool_call_id=tool_call["id"])
)
return {"messages": results}
避坑指南:务必保持tool_call_id与原始调用一致,这是关联请求与响应的关键。
4.3 条件判断逻辑
python复制from typing import Literal
def should_continue(state: MessagesState) -> Literal["tool_node", END]:
"""根据LLM输出决定下一步"""
last_msg = state["messages"][-1]
return "tool_node" if last_msg.tool_calls else END
这个函数形成了工作流的"大脑",控制着工具调用循环的启停。
5. 图工作流组装
5.1 构建计算图
python复制from langgraph.graph import StateGraph, START, END
builder = StateGraph(MessagesState)
# 添加节点
builder.add_node("llm_call", llm_call)
builder.add_node("tool_node", tool_node)
# 设置边关系
builder.add_edge(START, "llm_call")
builder.add_conditional_edges(
"llm_call",
should_continue,
{"tool_node": "tool_node", END: END}
)
builder.add_edge("tool_node", "llm_call")
# 编译最终代理
agent = builder.compile()
5.2 工作流可视化
在Jupyter中可以使用以下代码生成流程图:
python复制from IPython.display import Image
Image(agent.get_graph(xray=True).draw_mermaid_png())
图示会清晰展示:
- 初始LLM调用节点
- 条件判断分支
- 工具调用与返回路径
6. 代理测试与问题排查
6.1 基础测试案例
python复制from langchain.messages import HumanMessage
messages = [HumanMessage(content="计算(12乘以3)加(15除以5)的结果")]
response = agent.invoke({"messages": messages})
for msg in response["messages"]:
if msg.type == "human":
print(f"用户: {msg.content}")
elif msg.type == "ai":
print(f"AI: {msg.content}")
elif msg.type == "tool":
print(f"工具结果: {msg.content}")
预期执行流程:
- 解析用户问题
- 先调用乘法工具计算12*3
- 调用除法工具计算15/5
- 最后调用加法工具求和
- 返回最终结果
6.2 常见问题解决方案
问题1:工具未被正确识别
- 检查
@tool装饰器是否应用 - 确认工具文档字符串清晰完整
- 验证
bind_tools()调用是否成功
问题2:无限循环
- 检查
should_continue的条件逻辑 - 确保工具调用后返回了正确格式的ToolMessage
- 监控
llm_calls防止超过安全阈值
问题3:类型错误
- 在工具函数中添加类型验证
- 对除法等操作添加零除检查
- 使用pydantic进行输入校验
7. 扩展与优化方向
在实际使用中,我发现几个有价值的优化点:
- 批量处理支持:修改工具函数以支持批量输入,如
add([1,2,3]) - 复杂表达式解析:集成
eval安全计算字符串表达式 - 历史记忆:在state中添加conversation_history实现多轮对话
- 性能监控:通过
llm_calls统计实现自动化熔断机制
这个代理虽然现在只能处理基础运算,但框架设计已经具备了处理更复杂任务的潜力。比如添加单位换算、科学计算等工具,就能扩展为专业领域的计算助手。
