记忆与工具详解
Agent要实现复杂功能，需要两个关键组件：记忆（Memory）和工具（Tools）。本节深入讲解这两个核心概念。
Memory（记忆）
为什么需要记忆？
没有记忆的对话：
用户：我叫张三
AI：你好张三
用户：我叫什么？
AI：抱歉，我不知道你的名字
有记忆的对话：
用户：我叫张三
AI：你好张三
用户：我叫什么？
AI：你叫张三记忆类型
| 类型 | 说明 | 适用场景 |
|---|---|---|
| ConversationBuffer | 保存所有消息 | 短对话 |
| ConversationBufferWindow | 保留最近N轮 | 中等长度对话 |
| ConversationSummary | 自动总结 | 长对话 |
| VectorStoreMemory | 向量检索记忆 | 海量历史 |
实现记忆
方式一：使用LangChain Memory
python
from langchain.memory import ConversationBufferMemory
from langchain.agents import AgentExecutor
memory = ConversationBufferMemory(
memory_key="chat_history",
return_messages=True
)
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
memory=memory,
verbose=True
)方式二：手动管理历史（推荐）
python
from langchain_core.messages import HumanMessage, AIMessage
from langchain_core.chat_history import InMemoryChatMessageHistory
# 存储每个会话的历史
session_histories = {}
def get_history(session_id: str) -> InMemoryChatMessageHistory:
if session_id not in session_histories:
session_histories[session_id] = InMemoryChatMessageHistory()
return session_histories[session_id]
# 添加消息
history = get_history("user_123")
history.add_message(HumanMessage(content="你好"))
history.add_message(AIMessage(content="你好！有什么可以帮你的？"))
# 查看历史
print(history.messages)方式三：数据库持久化
python
from langchain_community.chat_message_histories import RedisChatMessageHistory
def get_redis_history(session_id: str):
return RedisChatMessageHistory(
session_id=session_id,
url="redis://localhost:6379"
)
# 使用Redis存储
chain_with_history = RunnableWithMessageHistory(
chain,
get_redis_history,
input_messages_key="input",
history_messages_key="chat_history"
)记忆窗口控制
python
from langchain.memory import ConversationBufferWindowMemory
# 只保留最近5轮对话
memory = ConversationBufferWindowMemory(
k=5,
memory_key="chat_history",
return_messages=True
)记忆总结
对于超长对话，可以使用LLM自动总结：
python
from langchain.memory import ConversationSummaryMemory
memory = ConversationSummaryMemory(
llm=model,
memory_key="chat_history",
return_messages=True
)
# LLM会自动总结早期对话，保留关键信息Tools（工具）
工具的本质
工具是一个可以被Agent调用的函数，包含：
- 名称：工具的标识
- 描述：告诉Agent这个工具做什么
- 参数Schema：定义输入参数
定义工具的三种方式
方式一：@tool装饰器（推荐）
python
from langchain_core.tools import tool
@tool
def search_web(query: str) -> str:
"""搜索互联网获取信息
Args:
query: 搜索关键词
Returns:
搜索结果摘要
"""
# 实际调用搜索API
import requests
response = requests.get(f"https://api.search.com?q={query}")
return response.json().get("summary", "未找到结果")
@tool
def send_email(to: str, subject: str, body: str) -> str:
"""发送邮件
Args:
to: 收件人邮箱
subject: 邮件主题
body: 邮件正文
"""
# 发送邮件逻辑
return f"邮件已发送至 {to}"方式二：Tool类
python
from langchain_core.tools import Tool
def multiply(a: int, b: int) -> int:
return a * b
multiply_tool = Tool(
name="multiply",
func=lambda x: multiply(**eval(x)),
description="两个数字相乘，输入格式：{'a': 数字, 'b': 数字}"
)方式三：StructuredTool
python
from langchain_core.tools import StructuredTool
from pydantic import BaseModel, Field
class CalculatorInput(BaseModel):
"""计算器输入参数"""
expression: str = Field(description="数学表达式")
def calculator(expression: str) -> str:
"""计算数学表达式"""
try:
return str(eval(expression))
except Exception as e:
return f"计算错误：{e}"
calculator_tool = StructuredTool.from_function(
func=calculator,
name="calculator",
description="计算数学表达式，如 '2 + 3 * 4'",
args_schema=CalculatorInput
)内置工具
LangChain提供了很多内置工具：
python
from langchain_community.tools import (
DuckDuckGoSearchRun, # 搜索
PythonREPLTool, # Python解释器
WikipediaQueryRun, # 维基百科
)
# 搜索工具
search = DuckDuckGoSearchRun()
print(search.invoke("Python"))
# Python解释器
python_repl = PythonREPLTool()
print(python_repl.invoke("print(2 + 2)"))工具调用流程
1. 用户提问："北京天气怎么样？"
2. Agent思考：需要调用get_weather工具
3. Agent构造参数：{"city": "北京"}
4. 执行工具：get_weather("北京")
5. 获取结果："晴天，25°C"
6. Agent生成回答工具返回类型
工具可以返回不同类型：
python
from langchain_core.tools import tool
from typing import Literal
@tool
def get_stock_price(symbol: str) -> dict:
"""获取股票价格"""
return {
"symbol": symbol,
"price": 150.25,
"change": 2.5
}
@tool
def analyze_sentiment(text: str) -> Literal["positive", "negative", "neutral"]:
"""分析情感"""
# 返回枚举值
if "好" in text or "喜欢" in text:
return "positive"
elif "差" in text or "讨厌" in text:
return "negative"
return "neutral"工具错误处理
python
@tool
def api_call(endpoint: str) -> str:
"""调用API"""
try:
response = requests.get(endpoint, timeout=5)
response.raise_for_status()
return response.text
except requests.Timeout:
return "错误：请求超时，请稍后重试"
except requests.RequestException as e:
return f"错误：{str(e)}"综合实战：智能助手
python
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import tool
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables import RunnableWithMessageHistory
# 定义工具
@tool
def get_weather(city: str) -> str:
"""获取城市天气"""
weathers = {"北京": "晴 25°C", "上海": "多云 28°C"}
return weathers.get(city, f"未找到{city}的天气")
@tool
def calculate(expression: str) -> str:
"""计算数学表达式"""
try:
return str(eval(expression))
except:
return "计算错误"
@tool
def get_time() -> str:
"""获取当前时间"""
from datetime import datetime
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
# 创建Agent
tools = [get_weather, calculate, get_time]
model = ChatOpenAI(model="gpt-4o")
prompt = ChatPromptTemplate.from_messages([
("system", "你是智能助手，使用工具帮助用户"),
("placeholder", "{chat_history}"),
("user", "{input}"),
("placeholder", "{agent_scratchpad}")
])
agent = create_tool_calling_agent(model, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools)
# 添加记忆
store = {}
def get_history(session_id):
if session_id not in store:
store[session_id] = InMemoryChatMessageHistory()
return store[session_id]
agent_with_history = RunnableWithMessageHistory(
agent_executor,
get_history,
input_messages_key="input",
history_messages_key="chat_history"
)
# 使用
config = {"configurable": {"session_id": "user_1"}}
agent_with_history.invoke(
{"input": "现在几点？"},
config=config
)
agent_with_history.invoke(
{"input": "北京天气怎么样？"},
config=config
)小结
| 组件 | 要点 |
|---|---|
| Memory | 持久化对话历史 |
| Tools | Agent的能力扩展 |
| @tool | 最简单的工具定义方式 |
| 错误处理 | 工具要优雅处理异常 |
下一步
学会了记忆和工具后，继续学习 RAG应用开发，让AI拥有知识库能力。