ARTICLE DETAIL

资讯详情

深耕网站建设、视觉设计与SEO优化的一线实战洞察。

AI应用成本优化实战:从Token机制到记忆管理,五大策略有效降低大模型API开销

AI应用成本优化实战:从Token机制到记忆管理,五大策略有效降低大模型API开销

1. 项目概述:当AI聊天变成“吞金兽”

最近在捣鼓各种AI应用和Agent项目时,我踩了一个不大不小的坑,差点让我的钱包“大出血”。事情是这样的,我搭建了一个基于大语言模型的对话Agent,想让它帮我处理一些日常的客服和问答任务。一开始跑得挺欢,直到我月底查看云服务账单,发现费用比预期高出了一大截。仔细一排查,问题就出在这个看似不起眼的“Token”上。

你可能也听说过“Token”,在AI圈子里,这玩意儿就是计费的核心单位。无论是调用OpenAI的API,还是使用国内外的各类大模型服务,最终的花销几乎都按消耗的Token数量来计算。我那个Agent,设计初衷是7x24小时待命,自动回复用户消息。但我忽略了一个关键点:我为了提升对话的连贯性和智能感,给Agent加上了“记忆”(Memory)功能。就是这个功能,让每一次简单的用户提问,背后都上演了一场“Token燃烧”的狂欢。

简单来说,用户发来一条消息,比如“帮我推荐周末去哪玩”。我的Agent在处理时,并不仅仅分析这一句话。它会从记忆库中调取和这个用户之前的所有聊天记录作为上下文,一起塞给大模型,以便做出更个性化的推荐。如果这个用户已经和我聊了10轮,那么模型实际处理的,就是这10轮历史对话加上新问题,总共11条消息的内容。这直接导致了单次API调用的Token消耗呈倍数级增长。原本可能只值1毛钱的交互,因为带上了沉重的“历史包袱”,成本轻松突破1块钱。如果用户量大、交互频繁,这种“偷偷扣钱”的效应会被无限放大,账单自然就不好看了。

这不仅仅是钱的问题。过长的上下文(即过多的Token)还会拖慢模型响应速度,甚至可能触及模型本身的上下文长度限制,导致请求失败。所以,无论是为了控制成本,还是为了提升应用性能和稳定性,学会给AI应用“降Token”,是每一个开发者、产品经理,甚至是普通用户都应该掌握的技能。这本手册,就是我花了真金白银买来的教训总结,希望能帮你省下那“价值千元”的学费。

2. Token机制深度解析:为什么你的钱烧得这么快?

要“降Token”,首先得明白Token是什么,以及大模型是如何根据它来计费的。很多人把Token简单理解为“单词”,这其实不准确,也容易导致误判。

2.1 Token的本质:不是单词,是“词元”

在自然语言处理中,尤其是像GPT这类基于Transformer架构的大模型,它们看到的并不是我们人类理解的单词或汉字,而是Token。你可以把Token理解为模型字典里的一个基本单位。对于英文,一个Token可能是一个短单词(如“cat”),也可能是一个长单词的一部分(如“unbelievable”可能会被拆成“un”、“believe”、“able”),甚至是一个标点符号。对于中文,情况更特殊:一个汉字通常就是一个Token,但一些常见词汇或成语也可能被合并为一个Token。

这种拆分方式(称为Tokenization)是由模型背后的“分词器”决定的。不同的模型(如GPT-3.5、GPT-4、Claude、国内的各种大模型)都有自己的一套分词规则和字典。这就导致同样一段中文文本,在不同的模型里,被切分成的Token数量可能有差异。

举个例子:“我喜欢编程”这句话。

  • 在某些分词器下,可能被切成['我', '喜欢', '编程']3个Token。
  • 在另一些分词器下,“编程”作为一个常见词,可能被视作一个整体,变成['我', '喜欢', '编程']还是3个Token,但如果“喜欢编程”是一个高频短语,甚至可能被切为['我', '喜欢编程']2个Token。

注意:千万不要用“字数”或“单词数”去估算Token!这是新手最容易犯的错误,也是预算失控的开始。一定要使用模型提供商官方提供的工具或库来进行精确计算。

2.2 计费逻辑:输入输出都算钱,上下文是隐形杀手

几乎所有主流的大模型API都采用类似的计费模式:输入Token + 输出Token = 总消耗Token

  • 输入Token:你发送给模型的全部内容。这包括:
    1. 系统提示词:你设定的AI角色、行为指令(如“你是一个专业的旅行顾问”)。
    2. 聊天历史:为了实现多轮对话(Memory),你附加上去的之前所有对话记录。
    3. 本次用户问题:用户当前提出的问题。
  • 输出Token:模型生成的回答内容。

这里的陷阱在于“聊天历史”。为了实现智能的、有记忆的对话(即AI Agent中的Memory模块),开发者必须将历史对话作为输入的一部分传递给模型。假设每轮对话平均消耗100个Token,那么:

  • 第1轮:输入100 + 输出100 = 200 Token
  • 第2轮:输入(历史100+新问题100)+ 输出100 = 300 Token
  • 第3轮:输入(历史200+新问题100)+ 输出100 = 400 Token
  • ...
  • 第10轮:输入(历史900+新问题100)+ 输出100 = 1100 Token

你看,到了第10轮,处理一次用户提问的成本(按Token量算)已经是第一轮的5.5倍了!如果你的应用有大量长对话用户,总成本就会像滚雪球一样增长。这就是标题里说的“每发1条消息,偷偷扣你「10倍」的钱”的核心原理——不是单价涨了,而是单次请求处理的“货物量”暴涨了。

2.3 Memory的代价:能力与成本的权衡

AI Agent中的Memory(记忆)是实现高级别自主性和个性化的关键。它通常分为几种类型:

  • 对话历史记忆:最直接的方式,存储原始的对话记录。
  • 摘要记忆:定期将长对话总结成一段摘要,后续只传递摘要而非全文。
  • 向量记忆:将对话内容转化为向量,存入数据库(如ChromaDB, Pinecone)。每次查询时,只检索与当前问题最相关的几条记忆。

第一种方式实现简单,但对Token的消耗是线性增长的,成本最高。第二种和第三种是典型的“降Token”策略,用额外的计算(摘要生成或向量检索)来换取每次API调用时输入Token的减少。这本手册后续的重点,就是教你如何聪明地实现后两种记忆方式,在保持Agent能力的同时,死死按住成本。

3. 实战:小白也能上手的五大降Token秘籍

理解了原理,我们直接上干货。下面这五个方法,从易到难,你可以根据自己项目的复杂度和技术能力进行选择和组合。

3.1 秘籍一:精简你的系统提示词

系统提示词是每次请求都必须发送的“固定成本”。很多新手喜欢写冗长、详细的提示词,仿佛写得越细AI就越听话。其实不然,提示词需要的是精准,而非冗长。

错误示范

“你是一个AI助手,你要乐于助人,要有创造力,要详细地回答用户的问题,但同时要保持简洁。你来自一个伟大的AI公司,你的知识截止到2023年10月。你不能回答有害的、非法的内容。你要用中文回答,语气要友好。如果用户问你不知道的,你要诚实地说不知道。另外,你还要注意……”

这种提示词充满了重复和模糊的形容词,消耗了大量Token却未必能提升效果。

优化技巧

  1. 删除冗余形容词:“乐于助人”、“有创造力”、“详细地”、“简洁的”——这些模型本身已经具备或在指令中隐含,可以删除或精简。
  2. 使用结构化指令:用编号、关键词让指令更清晰。
  3. 合并同类项:将行为准则、知识范围、回答格式分别归类陈述。

优化后示范

“角色:专业助手。知识截止:2023-10。规则:1.用中文回答。2.拒绝有害非法请求。3.不知则答不知。风格:直接、清晰。”

对比一下,前者可能超过50个Token,后者可能只有20个Token。对于海量请求,这个节省是巨大的。

实操心得:把系统提示词想象成给AI的“岗位说明书”,而不是“人物小传”。用最干练的语言明确核心职责和边界即可。可以准备几个不同版本进行A/B测试,在效果不变的前提下,选用Token最少的那个。

3.2 秘籍二:为历史对话做“摘要手术”

这是对抗长上下文消耗最有效的策略之一。与其把越来越长的原始对话历史全部塞给模型,不如定期将它们压缩成一段简短的摘要。

实现思路

  1. 设定摘要触发点:可以基于对话轮数(如每5轮)、Token数(如历史上下文超过500 Token时)或特定用户指令(如用户说“总结一下刚才说的”)。
  2. 调用模型生成摘要:当触发点到达时,将需要摘要的对话历史作为输入,请求模型生成一段简短的总结。提示词可以是:“请将以下对话内容总结成一段不超过100字的摘要,保留核心事实和决策。”
  3. 替换历史记录:用新生成的摘要,替换掉原有的那部分详细历史记录。后续的对话,都基于这个摘要和摘要之后的新对话进行。

示例流程: 假设用户在和旅行Agent规划行程:

用户1: 我想去上海玩。 AI1: 上海很棒,推荐外滩、迪士尼。计划几天? 用户2: 三天两晚。 AI2: 建议第一天外滩、南京路;第二天迪士尼;第三天豫园、新天地。预算多少? 用户3: 人均2000左右。

此时对话已3轮,触发摘要。将上面3轮对话发给模型,得到摘要:“用户计划上海三天两晚游,人均预算2000元。已建议行程:D1外滩/南京路,D2迪士尼,D3豫园/新天地。”

后续对话中,不再传递原始的3轮对话,只传递这个摘要作为“背景记忆”。当用户问:“那住宿有什么推荐吗?”,模型收到的输入将是:系统提示词 + 摘要(上海三天两晚游,预算2000...)+ 新问题(住宿推荐?)

这样一来,无论之前聊了多少,用于记录上下文的Token消耗被恒定地控制在一个很低的水平(摘要的长度)。

注意事项:摘要会损失细节。对于需要精确引用之前某句话的场景(如法律、医疗咨询),需谨慎使用,或采用“摘要+关键原文引用”的混合模式。

3.3 秘籍三:启用向量记忆库,实现精准检索

这是构建复杂AI Agent的黄金标准。它的核心思想是:不把所有记忆都塞进上下文,而是存到外部的数据库里,需要时只提取相关的

工作原理

  1. 存储阶段:将每一段对话(或用户提供的文档、知识)通过嵌入模型转化为一个高维向量(一堆数字),然后连同原文一起存入向量数据库(如Chroma, Weaviate, Pinecone)。
  2. 检索阶段:当用户提出新问题时,将这个问题也转化为向量。然后在向量数据库中,寻找与这个问题向量“最相似”的几段记忆(通常使用余弦相似度计算)。
  3. 注入上下文:只把这检索到的、最相关的几条记忆(比如3条),作为上下文插入到本次请求中,送给大模型。

优势

  • Token消耗可控:无论总记忆库有多大,每次注入上下文的条数是固定的,Token消耗稳定。
  • 相关性高:模型得到的都是和当前问题强相关的信息,回答质量更高。
  • 支持海量知识:可以将产品手册、FAQ、公司文档全部向量化,让Agent拥有“超强记忆力”。

简易实现步骤(以Python + ChromaDB为例)

# 1. 安装库 # pip install chromadb openai import chromadb from openai import OpenAI import hashlib # 初始化客户端和向量数据库 client = OpenAI(api_key="your-key") chroma_client = chromadb.PersistentClient(path="./memory_db") collection = chroma_client.get_or_create_collection(name="chat_memory") def store_memory(user_id, text): """存储一段对话记忆""" # 为这段文本生成向量 response = client.embeddings.create(model="text-embedding-3-small", input=text) embedding = response.data[0].embedding # 生成一个唯一ID(简易版) mem_id = hashlib.md5(f"{user_id}_{text}".encode()).hexdigest() # 存入数据库 collection.add( embeddings=[embedding], documents=[text], ids=[mem_id], metadatas=[{"user_id": user_id}] ) def get_relevant_memory(user_id, query, top_k=3): """检索与查询相关的记忆""" # 将查询语句也转化为向量 response = client.embeddings.create(model="text-embedding-3-small", input=query) query_embedding = response.data[0].embedding # 在数据库中进行相似度搜索 results = collection.query( query_embeddings=[query_embedding], n_results=top_k, where={"user_id": user_id} # 只检索该用户的记忆 ) if results and results['documents']: return "\n".join(results['documents'][0]) # 返回检索到的文本 return "" # 模拟使用 user = "user_123" # 用户每次说话后,存储记忆 store_memory(user, "用户说:我喜欢吃辣的和海鲜。") store_memory(user, "用户说:我预算有限,喜欢性价比高的餐厅。") # 当用户新提问时 new_question = "推荐一家上海的餐厅。" relevant_history = get_relevant_memory(user, new_question) # relevant_history 可能是:“用户说:我喜欢吃辣的和海鲜。\n用户说:我预算有限,喜欢性价比高的餐厅。” # 构建最终提示 final_prompt = f""" 相关历史记忆: {relevant_history} 当前问题:{new_question} 请根据用户的历史偏好回答问题。 """ # 然后将final_prompt发送给大模型

通过这种方式,无论你和AI聊了100句还是1000句,每次处理新问题时,模型看到的都只是通过智能检索筛选出来的、最相关的3-5句话,Token消耗被牢牢锁死。

3.4 秘籍四:设定清晰的上下文窗口与截断策略

你不能无限制地累积历史。必须为你的Agent设定一个“记忆边界”。

  • 硬性截断:这是最简单的策略。设定一个最大上下文Token数(例如2000 Token)。当新的请求加上历史记录超过这个限制时,从最旧的历史开始丢弃,直到总长度符合要求。这就像一个滑动窗口,只保留最近的一段对话。

    • 优点:实现简单,绝对可控。
    • 缺点:可能丢失重要的早期信息(比如用户一开始设定的核心需求)。
  • 智能截断:结合摘要和向量检索。当上下文快满时,不是简单丢弃最旧的内容,而是触发一个“摘要”动作,将较早的、完整的历史对话总结成一段摘要,然后用摘要替换掉那部分原始内容,腾出空间给新的对话。这相当于把详细的短期记忆,转化成了压缩的长期记忆。

实操配置建议: 对于大多数对话型Agent,我推荐采用混合策略

  1. 保留最近3-5轮完整的原始对话,保证对话的即时连贯性。
  2. 对于更早的对话,使用向量检索方式,只在需要时提取相关片段。
  3. 如果对话轮次非常多,可以定期(例如每天或每次会话开始时)用摘要来概括上一阶段的整体情况,然后清空或归档详细记录。

这样既保证了灵活性,又将单次请求的Token消耗控制在了可预测的范围内。

3.5 秘籍五:监控、分析与迭代优化

降Token不是一劳永逸的事情,你需要像关注服务器CPU一样关注Token消耗。

  1. 埋点监控:在每次调用大模型API时,记录下请求的Token数(输入/输出)、用户ID、时间戳和对话轮次。这些数据可以存入数据库或日志系统。
  2. 成本归因:通过用户ID,你可以分析出是哪些用户或哪些类型的对话最“烧钱”。是那些喜欢长聊的用户?还是处理复杂问题的会话?
  3. 制定优化目标:例如,设定“单次交互平均Token成本不超过X”或“长尾用户(对话>20轮)的Token增长率低于Y%”这样的目标。
  4. A/B测试:尝试不同的摘要策略、不同的向量检索条数(top_k)、不同的系统提示词,通过对比实验数据,找到效果与成本的最佳平衡点。

核心技巧:很多云服务商和API平台(如OpenAI)的控制台本身就提供了使用量分析。定期查看这些报表,关注“平均每请求Token数”和“Token消耗趋势图”,是发现成本异常的最快途径。

4. 不同场景下的降Token策略选型指南

掌握了核心方法,我们来看看如何将它们应用到具体的AI Agent场景中。没有一种策略是万能的,关键看你的需求。

4.1 场景一:智能客服机器人

特点:问题相对独立,但需要快速从知识库(FAQ、产品文档)中找到答案。对话可能不长,但知识库庞大。

核心挑战:如何将庞大的产品知识高效地提供给模型,而不至于每次请求都附带一本“书”。

推荐策略

  • 向量记忆库为主:将所有的产品文档、FAQ条目转化为向量存储。这是此类场景的标配。
  • 精简系统提示词:明确机器人的角色和回答格式(如“直接给出解决方案,引用知识库条目编号”)。
  • 实现步骤
    1. 预处理所有知识文档,切片成大小适中的段落(如每段200-500字)。
    2. 使用嵌入模型为每个段落生成向量,存入向量数据库。
    3. 用户提问时,将问题向量化,从数据库中检索最相关的3-5个段落。
    4. 将这些段落作为“参考知识”注入系统提示词,让模型基于此生成回答。
    5. 对话历史可采用简单轮次截断(如只保留最近2轮),因为客服对话的上下文依赖性相对较弱。

成本节省点:避免了将整个知识库作为上下文发送。每次请求只携带极少的、精准的相关知识片段,输入Token极大降低。

4.2 场景二:个性化学习伴侣/教练

特点:对话周期长,需要深刻理解用户长期的学习目标、进度和薄弱环节。记忆的连贯性和深度非常重要。

核心挑战:如何在长达数周或数月的交互中,维持对用户画像的精准记忆,而不让上下文无限膨胀。

推荐策略

  • 分层记忆系统:这是最复杂的,也是效果最好的。
    • 短期记忆:保留最近5-10轮完整对话,保证教学互动的自然流畅。
    • 长期记忆(向量库):将每次课的核心知识点、用户的练习成果、暴露出的问题点,以结构化的笔记形式存入向量数据库。例如:“[日期] 用户掌握了二元一次方程解法,但在应用题转化上仍有困难。”
    • 用户画像摘要:维护一份动态更新的用户摘要,例如:“高中生,数学基础中等,目标提升函数和几何板块,反应速度较快但粗心。” 这份摘要可以定期(如每周)根据长期记忆的内容由模型自动更新。
  • 工作流程
    1. 用户开启新会话。
    2. 系统加载“用户画像摘要”和从向量库中检索到的与本次学习主题相关的“长期记忆”片段。
    3. 结合“短期记忆”(本次会话的聊天记录)和上述信息,共同构成上下文,发送给模型。
    4. 会话结束后,将本次的核心内容总结后,更新到长期记忆向量库和用户画像摘要中。

成本节省点:用一份简短的“用户画像摘要”替代了冗长的历史行为描述。用按需检索的“长期记忆片段”替代了全部的学习历史。短期记忆则被限制在很小的窗口内。

4.3 场景三:自动化工作流Agent(如自动写周报、处理邮件)

特点:任务导向,输入可能是一份文档、一堆数据或一系列指令。需要准确理解任务要求,并访问相关数据源。

核心挑战:任务指令和待处理的数据可能很长,如何让模型准确理解而不超上下文限制。

推荐策略

  • 指令与数据分离:不要试图把100页的财报和复杂的处理指令一次性塞给模型。采用“分而治之”的策略。
  • 实现模式
    1. 任务规划层:先用一个简短的提示词让模型理解核心任务,并输出一个分步执行计划。例如:“请分析这份财报。第一步,提取营收和利润数据;第二步,计算同比增长率;第三步,总结主要风险点。”
    2. 步骤执行层:对于每一步计划,单独调用模型。每次调用只附加上该步骤所需的最小数据上下文。例如,执行“第一步”时,只发送财报中涉及“合并利润表”的那几页。
    3. 结果汇总层:最后,将各步骤的结果汇总,再让模型生成最终报告。
  • 利用函数调用:如果工作流固定,可以将某些步骤(如数据提取、计算)封装成函数,让模型通过“函数调用”来指挥外部工具执行,这比用自然语言描述所有操作更节省Token。

成本节省点:将一次处理超长文档的“重型”请求,拆分成多次处理局部数据的“轻型”请求。总Token消耗可能相近甚至更少,因为避免了将无关信息反复传递给模型,且每一步的指令更清晰,降低了模型的“思考”负担(输出Token)。

5. 高级技巧与避坑指南

当你掌握了基础方法后,下面这些进阶技巧和常见陷阱能帮你把优化做到极致。

5.1 Token估算与预算预警

在项目规划阶段,就必须对Token消耗进行估算。

  • 估算公式日均成本 ≈ 日均请求次数 × 平均每次请求Token数 × 每千Token单价
  • 如何获取“平均每次请求Token数”:在开发测试阶段,就要模拟真实用户对话流,记录下不同对话长度下的Token消耗,取一个保守的P90值(即90%的请求不会超过这个值)。
  • 设置预算警报:几乎所有云平台都支持设置预算和警报。务必设置一个每日/每月的消费上限和预警阈值(如达到80%时报警),这是防止“天价账单”的最后防线。

5.2 模型选择的成本考量

不同的模型,不仅能力不同,价格也天差地别。

  • 理解定价阶梯:通常,能力越强的模型(如GPT-4 Turbo),每千Token的价格越高。而一些较小的、优化的模型(如GPT-3.5-Turbo)价格则便宜得多。
  • 混合使用策略:可以采用“路由”策略。例如,让一个轻量级模型(或规则系统)先判断用户意图:如果是简单问答,用便宜模型;如果是需要复杂推理、创作或需要调用长期记忆的,再路由到强大但昂贵的模型。这样能用最低的成本覆盖大部分简单请求。

5.3 记忆管理中的常见陷阱

  1. 摘要的信息丢失:这是摘要策略最大的风险。自动化生成的摘要可能会遗漏用户认为重要的细节。缓解方法:对于关键决策点(如用户确认的订单信息、特殊要求),可以在生成摘要时,强制要求模型保留这些信息的原文引用。
  2. 向量检索的“幻觉”:向量检索是基于语义相似度,而不是精确匹配。有时会检索到看似相关实则不准确的记忆。缓解方法:在注入上下文时,可以加入元数据,如“来源置信度:高/中/低”,或让模型在回答时注明依据了哪条记忆。
  3. 记忆冲突与污染:在多人或多会话场景下,如果记忆存储没有做好隔离(如用唯一的用户ID、会话ID作为命名空间),可能会发生记忆串台,导致Agent对用户A说出了用户B的隐私。这是严重错误。务必在存储和检索时,严格使用user_idsession_id进行过滤。

5.4 性能与成本的平衡艺术

降Token的终极目标不是Token越少越好,而是在可接受的成本下,实现最佳的用户体验

  • 建立评估指标:除了成本,还要监控回答的准确率、用户满意度(如点赞/点踩率)、任务完成率等。
  • 进行灰度发布:当你引入一种新的降Token策略(如将摘要频率从5轮改为3轮),不要全量上线。先对小部分用户(如5%)进行灰度测试,对比实验组和对照组在成本与核心指标上的差异。
  • 接受合理的成本:有些场景就是需要消耗更多Token来保证质量。例如,法律合同审核Agent,必须看到完整的合同文本才能工作。这时,成本是提供该服务必须付出的价值,关键是通过定价将其转嫁给客户,而不是一味削减。

6. 工具链推荐与实战配置

工欲善其事,必先利其器。下面是我在项目中实际使用并验证过的一些工具和配置片段。

6.1 Token计算工具

  • OpenAI Tiktoken:OpenAI官方的分词器库,最准确。pip install tiktoken
    import tiktoken # 获取指定模型的编码器 enc = tiktoken.encoding_for_model("gpt-3.5-turbo") # 计算文本的Token数 text = "我喜欢编程和人工智能。" token_count = len(enc.encode(text)) print(f"Token数量: {token_count}") # 输出可能是 8(取决于分词规则)
  • 在线估算工具:一些第三方网站提供基于字符数的粗略估算,仅用于规划,不可用于精确计费。

6.2 向量数据库选型

数据库特点适用场景上手难度
ChromaDB轻量、开源、易集成,可直接在内存或磁盘运行,适合快速原型和中小项目。个人项目、初创公司、对运维要求不高的场景。
Weaviate功能强大,开源,自带向量化和模块化设计,支持混合搜索(关键词+向量)。需要高级检索功能、生产级应用。
Pinecone全托管云服务,无需运维,自动扩缩容,性能稳定。缺乏运维团队、需要快速搭建生产环境、预算充足。低(但需付费)
Qdrant开源,Rust编写,性能优异,Docker部署方便,提供云服务。对性能有高要求,希望在自托管和云服务间灵活选择。

个人建议:对于大多数AI Agent项目,从ChromaDB开始是最佳选择。它简单到几行代码就能跑起来,让你快速验证想法。当数据量变大、需要更高级功能时,再迁移到Weaviate或Qdrant。

6.3 一个完整的Agent记忆系统配置示例

以下是一个结合了短期记忆窗口、向量长期记忆和摘要的简化版Agent核心逻辑框架:

import chromadb from openai import OpenAI from datetime import datetime import json class SmartMemoryAgent: def __init__(self, user_id, llm_client, embed_client): self.user_id = user_id self.llm = llm_client self.embed_client = embed_client # 短期记忆:一个固定长度的列表,存储最近的原始对话 self.short_term_memory = [] # 格式:[{"role":"user", "content":"..."}, {"role":"assistant", "content":"..."}] self.SHORT_TERM_LIMIT = 6 # 保留最近3轮对话(user+assistant各一句为一轮) # 初始化向量数据库连接(长期记忆) self.chroma_client = chromadb.PersistentClient(path=f"./db/{user_id}") self.collection = self.chroma_client.get_or_create_collection(name="long_term_memory") # 用户摘要档案 self.user_profile = self._load_profile() def _load_profile(self): """加载或初始化用户摘要档案""" try: with open(f"./profiles/{self.user_id}.json", "r") as f: return json.load(f) except FileNotFoundError: return {"summary": "新用户,暂无历史信息。", "last_updated": None} def _save_profile(self): """保存用户摘要档案""" with open(f"./profiles/{self.user_id}.json", "w") as f: json.dump(self.user_profile, f) def _update_profile(self, conversation_segment): """根据一段对话,更新用户摘要(简化版:定期全量重写)""" # 这里可以设计更复杂的逻辑,比如每10轮对话触发一次 profile_prompt = f""" 以下是用户的最新对话片段: {conversation_segment} 结合之前的用户画像:{self.user_profile['summary']} 请更新用户画像摘要,用一段话概括用户的兴趣、需求或特点。保持简洁。 """ response = self.llm.chat.completions.create( model="gpt-3.5-turbo", messages=[{"role": "user", "content": profile_prompt}], max_tokens=150 ) new_summary = response.choices[0].message.content self.user_profile["summary"] = new_summary self.user_profile["last_updated"] = datetime.now().isoformat() self._save_profile() def _get_relevant_memories(self, query, top_k=2): """从向量库检索相关长期记忆""" # 生成查询向量 query_embed = self.embed_client.embeddings.create( model="text-embedding-3-small", input=query ).data[0].embedding results = self.collection.query( query_embeddings=[query_embed], n_results=top_k, where={"user_id": self.user_id} ) if results['documents']: return "相关历史记忆:\n" + "\n".join(results['documents'][0]) return "" def chat(self, user_input): """处理用户输入的核心方法""" # 1. 检索相关长期记忆 long_term_context = self._get_relevant_memories(user_input) # 2. 构建系统提示词,注入用户摘要和长期记忆 system_message = { "role": "system", "content": f"""你是一个智能助手。以下是对当前用户的简要了解: 用户画像:{self.user_profile['summary']} {long_term_context} 请基于以上信息,友好、专业地回应用户。""" } # 3. 构建本次请求的消息列表 messages = [system_message] # 加入短期记忆(最近的几轮对话) messages.extend(self.short_term_memory) # 加入用户当前输入 messages.append({"role": "user", "content": user_input}) # 4. 调用大模型 response = self.llm.chat.completions.create( model="gpt-3.5-turbo", messages=messages, max_tokens=500 ) ai_response = response.choices[0].message.content # 5. 更新短期记忆 self.short_term_memory.append({"role": "user", "content": user_input}) self.short_term_memory.append({"role": "assistant", "content": ai_response}) # 保持短期记忆不超过限制 if len(self.short_term_memory) > self.SHORT_TERM_LIMIT: self.short_term_memory = self.short_term_memory[-self.SHORT_TERM_LIMIT:] # 6. 将本轮有信息量的对话存入长期记忆(向量库) # 简单策略:只存储用户输入,或存储完整的Q-A对。这里存储Q-A对。 memory_text = f"用户:{user_input}\n助手:{ai_response}" memory_embed = self.embed_client.embeddings.create( model="text-embedding-3-small", input=memory_text ).data[0].embedding memory_id = f"{self.user_id}_{datetime.now().timestamp()}" self.collection.add( embeddings=[memory_embed], documents=[memory_text], ids=[memory_id], metadatas=[{"user_id": self.user_id, "type": "qa"}] ) # 7. 定期更新用户摘要(示例:每5轮对话更新一次) if len(self.collection.get()['ids']) % 5 == 0: # 简易触发条件 recent_conv = "\n".join([f"{m['role']}: {m['content']}" for m in self.short_term_memory[-6:]]) self._update_profile(recent_conv) return ai_response # 使用示例 openai_client = OpenAI(api_key="your-key") agent = SmartMemoryAgent(user_id="test_user_001", llm_client=openai_client, embed_client=openai_client) print(agent.chat("你好,我喜欢看电影和读书。")) print(agent.chat("能给我推荐一些科幻小说吗?"))

这个框架展示了如何将短期记忆、向量长期记忆和用户画像摘要组合在一起,形成一个既能记住关键信息,又能严格控制每次请求上下文长度的智能记忆系统。你可以根据自己的需求,调整各部分的触发条件和策略。

返回列表