从零开始:用 LangChain 构建带记忆的 AI 智能体(Agent)
从零开始:用 LangChain 构建带记忆的 AI 智能体(Agent)
前言
OpenAI 的函数调用(Function Calling)功能让大模型可以调用外部工具,而 LangChain 则把这一过程封装得更加优雅。今天,我将带你一步步构建一个带记忆功能的 AI Agent,让它不仅能调用工具查询天气,还能记住你和它的每一次对话。
最终效果如下:
🤖 Agent 已就绪,输入 '退出' 结束对话...
👤 你: 北京天气怎么样?
🤖 Agent: 北京当前天气晴朗,气温25°C,微风。
👤 你: 那上海呢?
🤖 Agent: 上海现在下着小雨,气温20°C,东南风。
👤 你: 我刚才问的第一个城市是哪里?
🤖 Agent: 您第一个问的是北京的天气情况。
整体架构
我们的 Agent 由三部分组成:
| 组件 | 作用 | 代码对应 |
|---|---|---|
| 大脑 | 理解用户意图,决定是否调用工具 | ChatOpenAI |
| 手脚 | 执行具体任务(如查天气) | @tool 装饰的函数 |
| 记忆 | 记住上下文和历史对话 | InMemorySaver |
环境准备
pip install langchain langchain-openai langgraph
如果你使用的是阿里云百炼平台(本代码示例使用 qwen3-max),还需要配置相应的 API Key。
完整代码解析
1. 初始化大模型
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(
model="qwen3-max", # 使用的模型名称
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1", # API 地址
api_key="your-api-key-here", # 替换为你的真实 API Key
temperature=0 # 温度越低,回答越确定
)
base_url指向阿里云百炼的兼容模式接口,支持 OpenAI 的调用格式temperature=0让模型输出更稳定,适合工具调用场景
2. 定义工具(Agent 的手脚)
from langchain_core.tools import tool
def get_weather(city: str) -> str:
"""获取指定城市的实时天气(这里用假数据模拟)"""
weather_data = {
"北京": "晴,气温 25°C,微风",
"上海": "小雨,气温 20°C,东南风",
"深圳": "多云,气温 28°C,微风"
}
return weather_data.get(city, f"抱歉,暂时查不到 {city} 的天气信息。")
关键点:
@tool装饰器会自动读取函数的 docstring 作为工具描述- 大模型就是根据这个描述来判断何时调用该工具的
- 参数的类型和注释也会被自动提取
3. 创建记忆模块
from langgraph.checkpoint.memory import InMemorySaver
memory = InMemorySaver()
InMemorySaver将对话历史保存在内存中- 程序重启后记忆会清空,适合开发和测试
- 生产环境可以替换为
RedisSaver或PostgresSaver
4. 组装 Agent
from langchain.agents import create_agent
agent = create_agent(
model=llm,
tools=[get_weather],
debug=True, # 开启调试,可以看到内部执行过程
checkpointer=memory # 传入记忆模块
)
create_agent是 LangChain 提供的高级 API,简化了 Agent 创建流程debug=True会打印出模型思考、工具调用等详细信息,非常有助于调试
5. 开启对话循环
config = {"configurable": {"thread_id": "chat-001"}}
print("🤖 Agent 已就绪,输入 '退出' 结束对话...")
while True:
user_input = input("\n👤 你: ")
if user_input.lower() in ["退出", "exit", "quit"]:
break
messages = {"messages": [("user", user_input)]}
response = agent.invoke(messages, config=config)
print(f"🤖 Agent: {response['messages'][-1].content}")
thread_id相当于会话 ID,同一个 ID 下的所有对话会被记忆模块串联起来- 不同的
thread_id之间的对话是隔离的
运行示例
下面是一个完整的对话流程:
🤖 Agent 已就绪,输入 '退出' 结束对话...
👤 你: 你好,我叫小明
🤖 Agent: 你好小明!有什么我可以帮你的吗?
👤 你: 北京今天天气如何?
🔧 正在调用工具: get_weather(city='北京')
🤖 Agent: 北京今天天气晴朗,气温25°C,微风。
👤 你: 那刚才那个城市有风吗?
🤖 Agent: 是的,北京今天有微风。您是问这个吧?
👤 你: 退出
进阶改进建议
1. 连接真实天气 API
import requests
def get_weather_real(city: str) -> str:
"""获取真实天气"""
api_key = "your-weather-api-key"
url = f"https://api.openweathermap.org/data/2.5/weather?q={city}&appid={api_key}&units=metric&lang=zh_cn"
# 处理请求逻辑...
2. 添加更多工具
def get_current_time() -> str:
"""获取当前时间"""
from datetime import datetime
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
def calculate(expression: str) -> str:
"""计算数学表达式,如 '1+2*3'"""
try:
return str(eval(expression))
except:
return "计算错误,请检查表达式"
3. 持久化记忆(生产环境)
from langgraph.checkpoint.sqlite import SqliteSaver
memory = SqliteSaver.from_conn_string("checkpoints.db")
常见问题
Q: 模型不调用工具怎么办?
- 检查工具函数的 docstring 是否清晰
- 降低
temperature参数 - 尝试更明确的提示词,如“请用工具查询”
Q: 如何查看 Agent 内部执行过程?
- 设置
debug=True - 或使用
langchain.callbacks.StdOutCallbackHandler
Q: 对话历史太长会不会超 Token?
- LangChain 的记忆模块会自动处理消息截断
- 也可以配置
max_tokens或使用trim_messages手动控制
总结
通过不到 50 行代码,我们搭建了一个具备记忆能力和工具调用能力的 AI Agent。这种模式可以轻松扩展到:
- 联网搜索
- 数据库查询
- 发送邮件
- 调用企业内部 API
Agent 的魅力在于:大模型负责“思考”,工具负责“执行”。当两者结合,AI 就不只是聊天机器人,而是真正的数字助手。