1. 为什么你的 Agent 总是“失忆”:用户画像记忆的真实困境
你有没有遇到过这种场景:上周刚跟电商 AI 客服说过自己对芒果过敏,这周再问零食推荐,它还是给你推芒果干;你在学习 App 上的 AI 助教已经知道你是 Python 入门水平,换到小程序端问编程问题,它又发来一堆计算机基础概念让你先学。这些问题的本质,是绝大多数 Agent 系统没有真正意义上的长期用户画像记忆能力——要么只靠有限的上下文窗口做短期会话记忆,要么全量存储会话历史导致 token 成本极高、上下文干扰严重。
用户画像记忆系统要解决的核心问题有三个:跨会话留存、低成本检索、精准注入。我试过把用户所有历史对话直接塞进 Prompt,结果单次请求 token 从 800 涨到 6000,响应时间翻了三倍,而且模型经常被无关历史带偏。后来改成“结构化标签 + 非结构化记忆片段”的混合方案,token 消耗降了七成以上,个性化匹配精度反而更高。
这篇文章要交付的,是一套可复制的个性化 Agent Harness 记忆链路:从用户画像的存储、检索到注入,配合 TaoToken 统一 Key/API 通道完成模型调用,最后给出验证画像记忆生效的测试动作。适合有一定 Python 基础、正在做 To C 类 AI 应用、被“千人一面”和 token 成本困扰的开发者。读完你能独立搭起一套生产级可用的画像记忆中间层,并知道每一步怎么验证、怎么排错。
2. TaoToken 前置准备:统一 Key 与 API 通道接入
在动手写记忆系统之前,先把模型调用通道理顺。个性化 Agent Harness 的记忆提炼、标签生成、上下文组装都依赖大模型,如果每个模块各自配置 Key,后期维护会很痛苦。TaoToken 提供统一的 API 通道,一个 Key 就能覆盖对话、编码、Agent 等多种模型调用场景,省去多平台切换的麻烦。
2.1 获取 API Key 与确认 Base URL
先到 TaoToken 控制台创建 API Key。访问 https://taotoken.net/api-keys 生成密钥,建议按项目维度创建,方便后续做用量隔离。拿到 Key 后,统一使用以下 Base URL:
Base URL: https://taotoken.net/api API Key: sk-你的密钥 Model ID: gpt-4o-mini(或你需要的模型)这三件套是后续所有配置的基础。无论你用的是 Claude Code、Cline、Codex 还是自己写的 Python 脚本,只要把 Base URL、Key、Model ID 填对,就能走通调用。如果你更习惯在对话界面里先验证模型是否可用,可以直接打开 https://taotoken.net/chat 发一条测试消息,确认通道正常再进入代码环节。
2.2 环境变量与依赖安装
把密钥写进环境变量,不要硬编码在代码里。在项目根目录创建.env文件:
TAOTOKEN_API_KEY=sk-你的密钥 TAOTOKEN_BASE_URL=https://taotoken.net/api TAOTOKEN_MODEL=gpt-4o-mini PG_URL=postgresql://postgres:123456@localhost:5432/agent_harness REDIS_URL=redis://localhost:6379/0安装依赖,这里用 OpenAI 兼容的 SDK 即可,因为 TaoToken 的 API 通道兼容标准接口:
pip install openai==1.13.3 chromadb==0.4.24 fastapi==0.109.2 \ uvicorn==0.27.1 sqlalchemy==2.0.27 redis==5.0.1 \ pydantic==2.6.1 python-dotenv==1.0.1 scikit-learn==1.4.1.post12.3 验证通道连通性
写一个最小脚本确认 Key 和 Base URL 可用:
import os from openai import OpenAI from dotenv import load_dotenv load_dotenv() client = OpenAI( api_key=os.getenv("TAOTOKEN_API_KEY"), base_url=os.getenv("TAOTOKEN_BASE_URL"), ) resp = client.chat.completions.create( model=os.getenv("TAOTOKEN_MODEL"), messages=[{"role": "user", "content": "回复两个字:正常"}], ) print(resp.choices[0].message.content)跑通后输出“正常”,说明通道没问题。这一步很关键,后面记忆提炼模块报错时,你能快速判断是通道问题还是业务逻辑问题。如果你打算长期做编码类 Agent,可以顺带了解 https://taotoken.net/coding-plan 的套餐,按用量规划成本更清晰。
3. 可复制配置:Agent Harness 记忆系统核心片段
这一节给出可直接复制的配置和代码片段,覆盖数据模型、记忆提炼、检索排序、上下文注入四个环节。路径和参数都按实际项目结构写,你照着改就能跑。
3.1 数据模型定义(models.py)
用 Pydantic 定义标签和记忆片段,字段设计决定了后续检索的灵活性:
from pydantic import BaseModel, Field from datetime import datetime from typing import List, Optional class ProfileTag(BaseModel): tag_id: Optional[str] = None user_id: str tag_key: str = Field(description="标签键,如 food_allergy") tag_value: str = Field(description="标签值,如 花生") confidence: float = Field(ge=0, le=1, description="置信度") valid_days: int = Field(default=3650, description="有效期天数") scene_id: Optional[str] = Field(default="global") create_time: Optional[datetime] = None class MemorySegment(BaseModel): memory_id: Optional[str] = None user_id: str content: str = Field(description="记忆内容文本") scene_id: Optional[str] = Field(default="global") embedding: Optional[List[float]] = None hit_count: int = Field(default=0) create_time: Optional[datetime] = None class ContextBuildRequest(BaseModel): user_id: str query: str scene_id: str = Field(default="global") top_k: int = Field(default=10) min_score: float = Field(default=0.3)3.2 记忆提炼模块(extractor.py)
用大模型从用户输入中提炼标签和记忆片段。Prompt 里加入标签体系约束,避免模型自由发挥导致标签泛滥:
import os, json from openai import OpenAI from dotenv import load_dotenv load_dotenv() class MemoryExtractor: def __init__(self): self.client = OpenAI( api_key=os.getenv("TAOTOKEN_API_KEY"), base_url=os.getenv("TAOTOKEN_BASE_URL"), ) self.model = os.getenv("TAOTOKEN_MODEL") def extract(self, user_id, scene_id, user_input, tag_schema): prompt = f"""你是用户记忆提炼专家。从用户输入中提炼标签和记忆片段。 标签键必须从以下体系选择:{json.dumps(tag_schema, ensure_ascii=False)} 明确提到的打1.0置信度,模糊提到打0.5。 临时偏好有效期30天,长期禁忌有效期3650天。 只输出JSON,格式:{{"tags": [...], "memories": [...]}} 用户输入:{user_input}""" resp = self.client.chat.completions.create( model=self.model, messages=[{"role": "user", "content": prompt}], temperature=0, ) try: return json.loads(resp.choices[0].message.content) except Exception as e: print(f"解析失败:{e}") return {"tags": [], "memories": []}3.3 检索与权重计算(retriever.py)
记忆重要度得分由四部分组成:语义相关性、命中频率、时间衰减、场景匹配。权重参数可按业务调整:
import numpy as np from datetime import datetime from sklearn.metrics.pairwise import cosine_similarity class MemoryRetriever: def __init__(self, storage, weights=None): self.storage = storage self.weights = weights or {"alpha": 0.4, "beta": 0.2, "gamma": 0.2, "delta": 0.2} self.lambda_decay = 0.01 def calculate_score(self, item, query_embedding, scene_id): item_embedding = self.storage.get_embedding( f"{item.tag_key}:{item.tag_value}" if hasattr(item, "tag_key") else item.content ) R = cosine_similarity([query_embedding], [item_embedding])[0][0] F = min(1.0, getattr(item, "hit_count", 1) / 10) days = (datetime.now() - item.create_time).days T = np.exp(-self.lambda_decay * days) C = 1.0 if item.scene_id == scene_id else 0.5 if item.scene_id == "global" else 0.0 return max(0.0, min(1.0, self.weights["alpha"] * R + self.weights["beta"] * F + self.weights["gamma"] * T + self.weights["delta"] * C))3.4 上下文组装与注入(harness.py)
把检索出的标签和记忆片段拼成结构化上下文,注入到 Agent 的 Prompt 中:
def build_context(self, request): query_embedding = self.storage.get_embedding(request.query) tags = self.storage.get_user_tags(request.user_id, request.scene_id) memories = self.storage.search_memory_segments( request.user_id, request.scene_id, query_embedding, request.top_k * 2) scored = [] for tag in tags: s = self.calculate_score(tag, query_embedding, request.scene_id) if s >= request.min_score and tag.confidence >= 0.3: scored.append((s, "tag", tag)) for mem in memories: s = self.calculate_score(mem, query_embedding, request.scene_id) if s >= request.min_score: scored.append((s, "memory", mem)) scored.sort(reverse=True, key=lambda x: x[0]) top = scored[:request.top_k] context = "【用户画像信息】\n" for _, kind, item in top: if kind == "tag": context += f"- {item.tag_key}: {item.tag_value}\n" else: context += f"- 记忆:{item.content}\n" return context如果你用 Claude Code 做开发,可以把这套 Harness 作为独立服务跑起来,然后在 Claude Code 里通过 HTTP 调用。相关配置参考 https://taotoken.net/claude-code-anthropic ,Base URL 和 Key 填法一致。
4. 验证请求:确认画像记忆真的生效
写完代码不等于记忆生效,必须用可复现的测试动作验证。下面给出三个层次的验证方法,从单元测试到端到端链路。
4.1 单元测试:标签提炼准确性
构造一条明确包含偏好的输入,检查提炼结果是否符合预期:
extractor = MemoryExtractor() result = extractor.extract( user_id="u_001", scene_id="ecommerce", user_input="我对花生过敏,平时喜欢喝无糖饮料", tag_schema={"food_allergy": "食物过敏", "preferred_drink": "偏好饮品"} ) print(result)预期输出应包含food_allergy: 花生和preferred_drink: 无糖饮料两个标签,置信度均为 1.0。如果模型返回了标签体系外的键,说明 Prompt 约束不够,需要加强。
4.2 检索测试:跨会话记忆命中
先写入一条记忆,再用相关 query 检索,确认能命中:
storage.save_tags([ProfileTag( user_id="u_001", tag_key="food_allergy", tag_value="花生", confidence=1.0, scene_id="ecommerce" )]) req = ContextBuildRequest( user_id="u_001", query="推荐一些零食", scene_id="ecommerce" ) context = harness.build_context(req) print(context)如果输出里包含food_allergy: 花生,说明检索链路通了。这里有个坑:如果min_score设得过高(比如 0.8),短 query 和标签的余弦相似度可能不够,导致命中失败。建议先用 0.3 起步,观察实际得分再调。
4.3 端到端测试:注入后模型行为变化
最关键的一步,验证注入画像后模型输出是否真的个性化:
def handle_request(user_id, query, scene_id="ecommerce"): context = harness.build_context(ContextBuildRequest( user_id=user_id, query=query, scene_id=scene_id)) full_prompt = f"{context}\n\n用户问题:{query}\n请结合用户画像回答。" resp = client.chat.completions.create( model=os.getenv("TAOTOKEN_MODEL"), messages=[{"role": "user", "content": full_prompt}], ) return resp.choices[0].message.content print(handle_request("u_001", "推荐一些零食"))对比两次调用:一次带画像上下文,一次不带。带画像时模型应主动避开花生类零食;不带时可能推荐花生酥。这个对比就是画像记忆生效的直接证据。实测下来,注入上下文后推荐准确率提升明显,但要注意上下文别超过 500 token,否则会挤占对话空间。
5. 常见报错排查:401、local proxy failed 与 choices 解析失败
接入过程中最容易卡在几个固定报错上,这里按真实错误信息给出排查路径。
5.1 401 Unauthorized
报错原文通常是Error code: 401 - {'error': {'message': 'Invalid API key'}}。原因有三类:Key 复制时带了空格、环境变量没加载、Base URL 写错。排查顺序:先确认.env文件在项目根目录且load_dotenv()在 import 之后调用;再打印os.getenv("TAOTOKEN_API_KEY")看是否为 None;最后确认 Base URL 是https://taotoken.net/api而不是带路径的完整地址。如果 Key 本身失效,到控制台重新生成一个。
5.2 local proxy failed / connection refused
报错APIConnectionError: Connection error或local proxy failed,通常是网络层问题。先确认本机能否访问https://taotoken.net/api,用 curl 测一下:
curl -X POST https://taotoken.net/api/chat/completions \ -H "Authorization: Bearer sk-你的密钥" \ -H "Content-Type: application/json" \ -d '{"model":"gpt-4o-mini","messages":[{"role":"user","content":"test"}]}'如果 curl 通但 Python 不通,检查是否有全局代理环境变量干扰,比如HTTP_PROXY、HTTPS_PROXY被设置成了不可用的地址。清掉这些变量再试。另外确认防火墙没有拦截 443 端口。
5.3 reading 'choices' 报错
报错KeyError: 'choices'或TypeError: 'NoneType' object is not subscriptable,说明 API 返回结构不符合预期。常见原因是模型名写错,比如把gpt-4o-mini写成gpt-4o_mini,服务端返回了错误信息而不是正常响应。排查方法:在调用后先打印完整响应:
resp = client.chat.completions.create(...) print(resp.model_dump_json(indent=2))看返回里有没有error字段。如果有,按错误信息修正模型名或参数。另一个原因是temperature等参数超出了模型支持范围,去掉多余参数再试。
5.4 OAuth 与鉴权配置错误
如果你用 Claude Code 或 Codex 这类工具接入,报错可能显示OAuth token invalid或auth.json not found。这类工具不走简单的 API Key,而是需要配置auth.json或环境变量。以 Codex 为例,在~/.codex/auth.json中填入:
{ "api_key": "sk-你的密钥", "base_url": "https://taotoken.net/api" }Claude Code 则在 settings 中配置ANTHROPIC_BASE_URL和ANTHROPIC_API_KEY,具体路径参考 https://taotoken.net/doc 。配置完重启工具,再跑一次验证请求。如果还报鉴权失败,检查配置文件权限是否为 600,避免被其他进程读取导致冲突。
6. 从记忆链路到长期 Agent:CTA 与下一步
走到这里,你已经有了一个能跑通的画像记忆链路:标签提炼、混合存储、权重检索、上下文注入,每一步都有验证方法和排错路径。接下来要做的,是把它接到真实的 Agent 场景里持续迭代。
如果你主要做对话类应用,想先验证模型在个性化上下文下的表现,可以直接在 https://taotoken.net/chat 里手动构造带画像的 Prompt,观察输出差异,确认效果后再落到代码。如果你在做长期编码类 Agent,需要稳定的模型调用和用量管理,https://taotoken.net/coding-plan 提供了按周期规划的方案,适合把 Harness 作为常驻服务跑。所有接入所需的 Key 和文档入口在 https://taotoken.net/api-keys 和 https://taotoken.net/doc ,配置时记得 Base URL 统一用https://taotoken.net/api。
最后分享一个实用技巧:记忆系统的效果不取决于代码多复杂,而取决于标签体系设计得是否贴合业务。先花时间把标签键定义清楚,再让模型往里填值,比让模型自由发挥再事后清洗要省力得多。上线后每周抽检一批低置信度标签,人工确认后回写,两三轮下来准确率就能稳定在可用区间。