拆解 Agent 核心原理｜从零动手实现简易 AI 智能体（九）

本文摘要历史记录无限增长会超出模型上下文窗口闲置会话也持续占用内存。本篇通过引入tiktoken进行精确 token 计量并实现基于消息数量的历史截断与会话超时清理机制。一、环境与前提本篇建立在第八篇完成的会话隔离之上。项目目录为agent-demo/入口文件main.py请确保其中已包含通过session_id和asyncio.Lock实现的sessions会话字典及event_generator函数。前置依赖Python 3.11已安装openai,fastapi,uvicorn本篇新增依赖tiktoken安装命令如下pipinstalltiktoken预期输出执行pip list | grep tiktoken将显示已安装的tiktoken包。实际输出tiktoken 0.5.1以实际安装版本为准。二、关键步骤步骤 1添加 Token 用量统计到ChatResponse目的返回每次请求消耗的 token 数便于成本监控。操作修改ChatResponse模型添加usage字段。在event_generator中流式响应结束后尝试从最后一个 chunk 提取usage信息并发送。# main.py 的相关修改部分frompydanticimportBaseModelfromtypingimportList,Dict,AnyclassChatResponse(BaseModel):reply:strusage:Dict[str,int]|NoneNoneasyncdefevent_generator(session_id:str,user_message:str,):# ... 会话获取、消息追加等前置逻辑 ...response_chunks[]usage_dataNoneasyncforchunkinopenai_stream:# ... 处理流式块追加文本到 response_chunks ...pass# 注意OpenAI 流式接口的 usage 提取方式取决于具体库版本以下为简化示例。# 更可靠的方式是使用非流式调用或确认 openai 库版本支持从流中获取 usage。ifhasattr(chunk,usage)andchunk.usage:usage_datachunk.usage.model_dump()final_reply.join(response_chunks)yieldfdata:{ChatResponse(replyfinal_reply,usageusage_data).json()}\n\n预期输出成功请求后SSE 流末尾的 JSON 数据将包含usage字段。实际输出未实测。步骤 2实现历史长度管理基于消息数量截断目的防止history列表无限增长导致 API 上下文超限。操作在event_generator函数中获取会话历史后、调用模型前对history进行截断。# main.py 的相关修改部分在 event_generator 函数内asyncdefevent_generator(session_id:str,user_message:str):asyncwithsessions_lock:session_datasessions.get(session_id,{history:[],last_active:None})# 更新最后活跃时间session_data[last_active]time.time()# 使用标准库 time# 添加当前用户消息session_data[history].append({role:user,content:user_message})# 截断历史保留最近 20 条消息MAX_HISTORY_MESSAGES20iflen(session_data[history])MAX_HISTORY_MESSAGES:session_data[history]session_data[history][-MAX_HISTORY_MESSAGES:]current_historysession_data[history].copy()# 后续使用 current_history 调用 openai_stream预期输出会话历史长度始终不超过 20 条。实际输出未实测。步骤 3实现会话超时清理目的定期清理长时间不活跃的会话释放内存。操作定义超时时间如 30 分钟。编写异步清理函数。在 FastAPI 的lifespan事件中启动清理任务。# main.py 的相关修改部分importasynciofromcontextlibimportasynccontextmanagerimporttime# 定义会话存储和锁沿用前文sessions:Dict[str,Dict]{}sessions_lockasyncio.Lock()SESSION_TIMEOUT30*60# 30 分钟单位秒asyncdefcleanup_expired_sessions():定期清理过期会话whileTrue:awaitasyncio.sleep(5*60)# 每 5 分钟检查一次current_timetime.time()asyncwithsessions_lock:expired_sessions[sidforsid,datainsessions.items()ifcurrent_time-data.get(last_active,current_time)SESSION_TIMEOUT]forsidinexpired_sessions:delsessions[sid]print(f[Cleanup] Session{sid}expired and removed.)asynccontextmanagerasyncdeflifespan(app:FastAPI):cleanup_taskasyncio.create_task(cleanup_expired_sessions())yieldcleanup_task.cancel()try:awaitcleanup_taskexceptasyncio.CancelledError:passappFastAPI(lifespanlifespan)预期输出服务器运行期间超过 30 分钟无活动的会话将被自动删除控制台有清理日志。实际输出未实测。步骤 4整合并更新run_agent_stream调用可选增强目的通过max_tokens参数限制模型单次回复长度。操作在构建openai.chat.completions.create调用时添加max_tokens参数。# 在 event_generator 函数中openai_streamawaitopenai.chat.completions.create(modelgpt-3.5-turbo,messagescurrent_history,streamTrue,max_tokens1024# 限制单次回复最大 token 数)预期输出模型单次回复长度受max_tokens1024限制。实际输出未实测。三、失败处理常见失败使用tiktoken.encoding_for_model(gpt-3.5-turbo)时可能因模型映射缺失而报错。报错原文KeyError: Could not automatically map gpt-3.5-turbo to a tokeniser. Please use tiktoken.get_encoding to explicitly get the tokeniser you expect.原因tiktoken库版本可能未包含最新模型的分词器映射。修复方法直接指定编码名称。importtiktoken encodingtiktoken.get_encoding(cl100k_base)# 适用于 gpt-3.5-turbo 和 gpt-4token_countlen(encoding.encode(text))四、替代方案与取舍本文演示了基于消息数量的截断但还有其他方案。做法适用条件代价边界不适用场景基于消息数量的截断1. 原型开发快速验证。2. 对话轮数固定且较短。1. 实现简单。2. 无法精确控制总 token 消耗可能浪费额度或意外超限。适用于对话结构稳定、每轮交互 token 消耗差异不大的场景。当用户问题或模型回复长度波动极大时容易导致总 token 数超限。基于 token 总量的截断1. 生产环境需严格控制成本与 API 调用成功率。2. 使用多种模型需精确管理上下文窗口。1. 需引入tiktoken依赖。2. 每次截断前需计算 token 数增加延迟和 CPU 开销。适用于任何需要精确控制 token 的场景是更健壮的方案。对于极简单的演示项目可能略显过度设计。选择建议生产环境中尤其是处理长文档或长对话时建议结合tiktoken实现基于 token 总量的截断。参考资料Background Tasks - FastAPIasyncio.sleep — Python 官方文档OpenAI API 文档 - Chat Completionstiktoken GitHub 仓库下一篇将探讨如何将对话历史与 Agent 状态持久化到数据库实现跨服务器重启的会话恢复。