共计 3449 个字符，预计需要花费 9 分钟才能阅读完成。
痛点分析
开发者在构建 AI Agent 时常常遇到以下问题：
- 架构设计模糊 ：缺乏清晰的模块划分，导致功能耦合度高，后期维护困难。
- 工具链复杂 ：集成多种工具（如数据库、API 服务）时，配置繁琐且易出错。
- 状态管理困难 ：对话上下文丢失，尤其是在多轮对话中，难以保持连贯性。
- 工具调用不稳定 ：外部 API 调用失败或超时，缺乏有效的重试和容错机制。
- 性能瓶颈 ：对话延迟高，用户体验差，尤其在处理复杂任务时。
技术选型
主流 AI Agent 开发框架对比：
| 框架 | 开发效率 | 扩展性 | 社区支持 | 适用场景 |
|---|---|---|---|---|
| LangChain | 高 | 强 | 活跃 | 复杂对话流、工具集成 |
| Semantic Kernel | 中 | 中 | 一般 | 微软生态、简单任务 |
| Rasa | 低 | 强 | 活跃 | 自定义 NLU、复杂对话管理 |
选择 LangChain 的核心理由：
- 模块化设计 ：Chain 式调用清晰，易于组合功能。
- 丰富的工具集成 ：内置多种工具（如搜索引擎、计算器）。
- 强大的记忆机制 ：支持多种记忆后端（Redis、数据库等）。
- 活跃的社区 ：问题解决速度快，案例丰富。
核心实现
基于 LangChain 的对话管理模块
from langchain.chains import ConversationChain
from langchain.memory import ConversationBufferMemory
# 初始化带历史缓存的 ConversationChain
memory = ConversationBufferMemory() # 默认存储最近的对话历史
conversation = ConversationChain(
llm=your_llm_model, # 替换为实际的 LLM 模型
memory=memory,
verbose=True # 打印详细日志
)
# 示例对话
response = conversation.predict(input="你好，我是开发者")
print(response) # 输出模型回复 异常处理装饰器实现
import functools
import time
from typing import Callable
def retry_on_failure(max_retries: int = 3, delay: float = 1.0):
"""工具调用失败时自动重试的装饰器"""
def decorator(func: Callable):
@functools.wraps(func)
def wrapper(*args, **kwargs):
retries = 0
while retries < max_retries:
try:
return func(*args, **kwargs)
except Exception as e:
print(f"调用失败: {e}, 重试 {retries + 1}/{max_retries}")
retries += 1
time.sleep(delay)
raise Exception(f"超过最大重试次数 {max_retries}")
return wrapper
return decorator
# 使用示例
@retry_on_failure()
def call_weather_api(city: str):
# 模拟可能失败的 API 调用
if "error" in city.lower():
raise ValueError("无效城市名")
return f"{city} 天气: 晴朗"工具调用的幂等性保证
def idempotent_tool_call(tool_func: Callable, *args, **kwargs):
"""确保工具调用幂等性的封装函数"""
call_id = hash(frozenset(kwargs.items())) # 生成唯一调用 ID
cached_result = cache.get(call_id) # 假设有缓存系统
if cached_result:
return cached_result
result = tool_func(*args, **kwargs)
cache.set(call_id, result, ttl=300) # 缓存 5 分钟
return result生产级考量
对话延迟优化方案
-
异步处理 ：
import asyncio from langchain.chains import LLMChain async def async_predict(chain: LLMChain, input_str: str): """异步执行预测减少等待时间""" return await chain.arun(input=input_str) -
流式响应 ：
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler # 初始化时添加流式回调 streaming_llm = LLM(callbacks=[StreamingStdOutCallbackHandler()], streaming=True )
敏感信息过滤
import re
SENSITIVE_PATTERNS = [r"\b\d{4}[-]?\d{4}[-]?\d{4}\b", # 信用卡号
r"\b\d{3}-?\d{2}-?\d{4}\b", # SSN
r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b" # 邮箱
]
def filter_sensitive(text: str) -> str:
"""使用正则表达式过滤敏感信息"""
for pattern in SENSITIVE_PATTERNS:
text = re.sub(pattern, "[REDACTED]", text)
return text避坑指南
避免记忆泄露
# 设置 TTL 的 Redis 记忆后端
from langchain.memory import RedisChatMessageHistory
message_history = RedisChatMessageHistory(
redis_url="redis://localhost:6379/0",
session_id="user123",
ttl=3600 # 1 小时后自动过期
)工具调用超时重试
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=4, max=10)
)
def reliable_api_call():
# 实现带指数退避的重试逻辑
pass对话状态持久化方案对比
| 方案 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|
| 内存 | 零延迟 | 重启丢失 | 开发环境 |
| Redis | 高性能，支持 TTL | 需要额外基础设施 | 生产环境 |
| 数据库 | 持久化可靠 | 读写性能较低 | 关键任务数据 |
| 文件系统 | 简单易用 | 并发访问困难 | 单机小规模应用 |
实践任务
实现一个支持天气查询 + 备忘录的复合型 Agent：
-
初始化 LangChain 环境
pip install langchain openai python-dotenv -
创建基础 Agent 类
from langchain.agents import Tool, AgentExecutor from langchain.agents import initialize_agent tools = [ Tool( name="Weather", func=get_weather, description="查询城市天气" ), Tool( name="Memo", func=save_memo, description="保存备忘录" ) ] agent = initialize_agent( tools, llm, agent="conversational-react-description", verbose=True ) -
实现工具函数
def get_weather(city: str) -> str: """调用天气 API（模拟实现）""" return f"{city} 天气: 25°C 晴天" memos = [] def save_memo(content: str) -> str: """保存备忘录到内存""" memos.append(content) return f"已保存备忘录: {content}" -
测试复合功能
agent.run("北京天气怎么样？") agent.run("记住下午 3 点有会议") agent.run("我刚才让你记了什么？") # 需要实现查询功能
完成以上步骤后，一个基础的复合型 Agent 就构建成功了。建议进一步添加异常处理、记忆持久化和用户身份验证等功能。
正文完
