1. 为什么我要造一个「越用越懂你」的个人智能体先说结论市面上大部分所谓的个人智能体本质上只是一个套了壳的聊天窗口。你问一句它答一句关掉页面之后它对你一无所知下次打开还是从零开始。这种体验用一句话概括就是——它认识你但记不住你。我做了三年多的后端和 AI 应用开发从最早的规则引擎客服到后来的 RAG 知识库再到这两年各种 Agent 框架轮番上阵踩过的坑能写一本书。去年年底我开始认真思考一个问题如果我要给自己造一个真正意义上的「数字分身」它应该长什么样不是那种演示用的玩具而是每天真的能帮我处理信息、记住我的偏好、随着使用越来越顺手的东西。这个系列我打算完整记录这套超体技术架构的设计和落地过程。所谓「超体」是我给这套架构起的名字核心思路是三层感知层负责接住输入认知层负责理解和记忆行动层负责真正干活。整套系统基于 FastAPI 做服务骨架用 DeepSeek 作为主力推理模型配合一套自己设计的记忆机制让智能体在长期交互中逐渐形成对使用者的「画像」。这篇文章是系列的开篇重点讲清楚三件事这套架构整体是怎么设计的、每个模块为什么这么选、以及一个最小可运行版本怎么搭起来。适合有一定 Python 基础、想从「调 API」进阶到「做系统」的开发者。如果你只是想知道智能体是什么网上科普文一抓一大把但如果你想真正动手做一个能长期用下去的个人智能体那接下来的内容应该对你有用。我先把话说在前面这套架构不是最优解甚至在某些场景下显得有点「重」。但它是我在实际使用中反复调整后留下来的方案每一个模块的存在都有具体的理由。我会把「为什么这么选」讲透而不是甩一堆代码让你自己猜。2. 超体架构的整体设计与模块拆解2.1 三层架构的核心思路很多人做智能体上来就纠结用哪个框架。LangChain、LangGraph、AutoGen、CrewAI名字一个比一个唬人。我的建议是先想清楚你的智能体要解决什么问题再决定用什么工具。框架是手段不是目的。超体架构的三层划分对应的是三个根本问题感知层用户说了什么以什么形式说的是文字、语音还是某个系统推送的事件认知层这句话是什么意思和之前的对话有什么关系需要调用哪些记忆行动层基于理解应该做什么是直接回答还是调用工具还是触发某个流程这三层听起来像是废话但真正落地的时候很多人的代码是混在一起的——路由里既做参数校验又做意图识别还顺手把数据库查了。这种写法在 demo 阶段没问题一旦要加功能就会变成一团乱麻。我选择用 FastAPI 作为整个系统的骨架原因很直接异步支持好、类型提示友好、自动生成文档。智能体系统天然是 IO 密集型的一次对话可能要等模型推理、等数据库查询、等外部 API 返回同步框架在这种场景下就是灾难。FastAPI 基于 Starlette 的异步能力配合 Pydantic 做数据校验写起来非常舒服。提示如果你之前只用过 Flask 或 Django转 FastAPI 最大的心智转变是「一切皆 async」。但要注意不是所有库都支持异步遇到同步阻塞的库要用run_in_executor包一层否则会卡住整个事件循环。2.2 为什么选 DeepSeek 作为主力模型模型选型这块我纠结了很久。早期用过一些闭源 API效果确实好但成本和数据隐私是绕不过去的坎。个人智能体要处理大量私人信息——日程、笔记、聊天记录这些东西交给第三方总归不太放心。DeepSeek 吸引我的点有三个推理能力强、API 兼容 OpenAI 格式、成本可控。尤其是它的推理能力在处理需要多步思考的任务时表现明显好于同价位的模型。我实测过一个场景让它根据我一周的日程和待办自动规划出下周的时间安排同时考虑通勤、会议冲突和个人休息时间。这种任务需要模型理解约束、做权衡、给出可解释的方案DeepSeek 的完成度让我比较满意。API 兼容 OpenAI 格式这点也很关键。意味着我可以直接用openai这个 Python 库只需要改base_url和api_key代码几乎不用动。这在我做模型对比测试的时候省了大量时间。from openai import AsyncOpenAI client AsyncOpenAI( api_keyyour-api-key, base_urlhttps://api.deepseek.com/v1 ) async def chat(messages: list[dict]) - str: response await client.chat.completions.create( modeldeepseek-chat, messagesmessages, temperature0.7, max_tokens2048 ) return response.choices[0].message.content这段代码就是最基础的调用封装。但注意直接这样用是不够的因为每次调用都是无状态的模型不记得上一轮说了什么。要让它「记得住」就得在messages里把历史对话带上。而历史对话怎么存、存多少、怎么检索就是认知层要解决的问题。2.3 记忆机制让智能体真正「懂你」的关键这是整套架构里我最花心思的部分。市面上大部分教程讲智能体记忆这块要么一笔带过要么就是简单地把对话历史塞进 context。但真正的「越用越懂你」需要的是分层记忆。我把记忆分成三类记忆类型存储内容存储方式生命周期短期记忆当前会话的对话历史内存/Redis会话结束即清除长期记忆用户偏好、重要事实向量数据库永久工作记忆当前任务的中间状态内存任务结束即清除短期记忆好理解就是最近几轮对话。但这里有个坑不能无限制地往 context 里塞历史模型有 token 上限塞太多不仅贵还会导致模型「注意力涣散」反而记不住重点。我的做法是保留最近 N 轮完整对话更早的对话做摘要压缩。长期记忆是「懂你」的核心。比如你告诉过它「我不吃香菜」「我习惯早上七点起床」「我的项目代号叫超体」这些信息应该被提取出来存进向量数据库在后续对话中按需检索。这里我用的是语义检索 关键词检索的混合方案因为纯语义检索有时候会漏掉精确匹配的信息。工作记忆则是为了支持多步任务。比如智能体在帮你规划行程时需要记住「已经查了航班」「还没订酒店」这些中间状态。这部分我暂时用内存字典实现简单够用。注意记忆的写入时机很关键。不要每轮对话都往长期记忆里写那样会引入大量噪音。我的策略是让模型自己判断「这条信息是否值得长期记住」通过一个单独的提取步骤来完成。2.4 目录结构设计FastAPI 项目的目录结构直接影响后续的可维护性。我见过太多项目把所有路由塞在一个main.py里超过五百行之后就没法看了。超体架构的目录结构是这样的chaoti/ ├── app/ │ ├── __init__.py │ ├── main.py # 应用入口 │ ├── config.py # 配置管理 │ ├── api/ │ │ ├── __init__.py │ │ ├── chat.py # 对话相关路由 │ │ ├── memory.py # 记忆管理路由 │ │ └── health.py # 健康检查 │ ├── core/ │ │ ├── __init__.py │ │ ├── agent.py # 智能体核心逻辑 │ │ ├── memory.py # 记忆管理 │ │ └── tools.py # 工具注册与调用 │ ├── models/ │ │ ├── __init__.py │ │ ├── schemas.py # Pydantic 模型 │ │ └── database.py # 数据库模型 │ ├── services/ │ │ ├── __init__.py │ │ ├── llm.py # 模型调用封装 │ │ └── embedding.py # 向量化服务 │ └── utils/ │ ├── __init__.py │ └── logger.py # 日志配置 ├── tests/ ├── requirements.txt └── .env这个结构的好处是职责清晰。api层只负责接收请求和返回响应业务逻辑在core和services里数据模型在models里。想加一个新功能你知道该往哪个目录放。3. 核心模块的实操落地3.1 环境准备与依赖安装先把基础环境搭起来。我假设你用的是 Python 3.10 以上因为要用到一些新的类型语法。python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate pip install fastapi uvicorn openai pydantic pydantic-settings pip install chromadb # 向量数据库轻量级够用 pip install python-dotenv这里解释几个关键依赖的选择理由uvicornASGI 服务器FastAPI 的标配。生产环境可以配合 gunicorn 用多 worker 模式。chromadb向量数据库我选的是 Chroma原因是它支持本地持久化、API 简单、不需要额外部署服务。如果你数据量特别大可以考虑 Milvus 或 Qdrant但个人智能体这个量级Chroma 完全够用。pydantic-settings管理配置比直接读环境变量优雅得多。提示Chroma 在 Windows 上偶尔会有编译问题如果装不上可以试试pip install chromadb --no-deps然后手动装依赖。或者直接用 Docker 跑一个 Chroma 服务。配置管理我用pydantic-settings来做把 API key、数据库路径这些放在.env文件里# app/config.py from pydantic_settings import BaseSettings class Settings(BaseSettings): deepseek_api_key: str deepseek_base_url: str https://api.deepseek.com/v1 model_name: str deepseek-chat chroma_path: str ./data/chroma max_history_rounds: int 10 class Config: env_file .env settings Settings()这样在代码里直接用settings.deepseek_api_key就行类型安全还有自动补全。3.2 对话接口的实现对话接口是整个系统的入口我把它设计成流式返回因为等待模型完整生成再返回的体验太差了。FastAPI 支持StreamingResponse配合模型的流式输出可以实现打字机效果。# app/api/chat.py from fastapi import APIRouter from fastapi.responses import StreamingResponse from app.models.schemas import ChatRequest from app.core.agent import ChaotiAgent router APIRouter(prefix/api/chat, tags[chat]) agent ChaotiAgent() router.post(/stream) async def chat_stream(request: ChatRequest): async def event_generator(): async for chunk in agent.stream_chat( user_idrequest.user_id, messagerequest.message, session_idrequest.session_id ): yield fdata: {chunk}\n\n return StreamingResponse( event_generator(), media_typetext/event-stream )这里用的是 SSEServer-Sent Events协议比 WebSocket 简单对于单向的流式输出场景足够用。前端用EventSource就能接。请求体的 Pydantic 模型# app/models/schemas.py from pydantic import BaseModel, Field class ChatRequest(BaseModel): user_id: str Field(..., description用户唯一标识) session_id: str Field(..., description会话标识) message: str Field(..., min_length1, max_length4000)user_id和session_id分开是有讲究的。user_id标识「谁」session_id标识「哪次对话」。长期记忆绑定在user_id上短期记忆绑定在session_id上。这样同一个用户开多个会话长期记忆是共享的但短期上下文互不干扰。3.3 智能体核心逻辑ChaotiAgent是整个系统的大脑它要协调记忆检索、模型调用、工具执行这几件事。我把它写成一个类核心方法是stream_chat# app/core/agent.py from app.services.llm import LLMService from app.core.memory import MemoryManager from app.core.tools import ToolRegistry class ChaotiAgent: def __init__(self): self.llm LLMService() self.memory MemoryManager() self.tools ToolRegistry() async def stream_chat(self, user_id: str, message: str, session_id: str): # 1. 检索长期记忆 long_term await self.memory.retrieve_long_term( user_iduser_id, querymessage, top_k5 ) # 2. 获取短期记忆 short_term await self.memory.get_short_term(session_id) # 3. 组装 prompt messages self._build_messages( long_termlong_term, short_termshort_term, user_messagemessage ) # 4. 流式调用模型 full_response async for chunk in self.llm.stream(messages): full_response chunk yield chunk # 5. 更新记忆 await self.memory.append_short_term( session_id, user, message ) await self.memory.append_short_term( session_id, assistant, full_response ) # 6. 异步提取长期记忆不阻塞响应 await self.memory.extract_and_store( user_iduser_id, conversationf用户: {message}\n助手: {full_response} )这个流程看起来简单但每一步都有细节。比如第 1 步的长期记忆检索不是简单地把所有记忆都拉出来而是根据当前消息做语义检索只取最相关的几条。第 6 步的长期记忆提取我用了一个单独的模型调用来判断「这段对话里有没有值得长期记住的信息」。_build_messages方法负责组装最终的 prompt这里有个技巧把长期记忆放在 system prompt 里短期记忆作为对话历史。这样模型能清楚区分「关于用户的背景知识」和「当前对话的上下文」。def _build_messages(self, long_term, short_term, user_message): system_prompt 你是超体一个个人智能体。 你需要根据以下关于用户的背景信息来提供个性化服务 {memory_context} 要求 1. 回答要自然不要生硬地复述背景信息 2. 如果背景信息与当前问题无关忽略即可 3. 保持友好、专业的语气 memory_text \n.join([f- {m} for m in long_term]) if long_term else 暂无背景信息 messages [ {role: system, content: system_prompt.format(memory_contextmemory_text)} ] messages.extend(short_term) messages.append({role: user, content: user_message}) return messages3.4 记忆管理的实现细节记忆管理是超体架构里最复杂的部分我拆成三个子模块来讲。短期记忆用内存字典加过期时间实现简单高效# app/core/memory.py from collections import defaultdict from datetime import datetime, timedelta class ShortTermMemory: def __init__(self, max_rounds: int 10, ttl_hours: int 24): self.sessions defaultdict(list) self.max_rounds max_rounds self.ttl timedelta(hoursttl_hours) self.last_access {} async def get(self, session_id: str) - list[dict]: self._cleanup() self.last_access[session_id] datetime.now() return self.sessions.get(session_id, []) async def append(self, session_id: str, role: str, content: str): self.sessions[session_id].append({ role: role, content: content, timestamp: datetime.now().isoformat() }) # 超过最大轮数时保留最近的 if len(self.sessions[session_id]) self.max_rounds * 2: self.sessions[session_id] self.sessions[session_id][-self.max_rounds * 2:] def _cleanup(self): now datetime.now() expired [ sid for sid, t in self.last_access.items() if now - t self.ttl ] for sid in expired: self.sessions.pop(sid, None) self.last_access.pop(sid, None)注意max_rounds * 2这个细节因为一轮对话包含 user 和 assistant 两条消息所以要乘 2。这个坑我踩过一开始只保留 10 条消息结果发现只有 5 轮对话上下文严重不足。长期记忆用 Chroma 做向量存储import chromadb from chromadb.config import Settings as ChromaSettings class LongTermMemory: def __init__(self, path: str): self.client chromadb.PersistentClient( pathpath, settingsChromaSettings(anonymized_telemetryFalse) ) self.collection self.client.get_or_create_collection( nameuser_memory, metadata{hnsw:space: cosine} ) async def store(self, user_id: str, content: str, metadata: dict None): import uuid doc_id str(uuid.uuid4()) self.collection.add( ids[doc_id], documents[content], metadatas[{user_id: user_id, **(metadata or {})}] ) async def retrieve(self, user_id: str, query: str, top_k: int 5): results self.collection.query( query_texts[query], n_resultstop_k, where{user_id: user_id} ) if not results[documents]: return [] return results[documents][0]这里用where{user_id: user_id}做过滤确保只检索当前用户的记忆。多用户场景下这个过滤是必须的否则会串数据。记忆提取是让智能体「自己判断什么值得记」的关键。我用一个单独的 prompt 让模型做这件事EXTRACT_PROMPT 分析以下对话提取出关于用户的、值得长期记住的信息。 对话内容 {conversation} 提取规则 1. 只提取关于用户的稳定信息偏好、习惯、事实 2. 不要提取一次性的、临时的信息 3. 每条信息独立成句简洁明了 4. 如果没有值得记住的信息返回空 以 JSON 数组格式返回例如[用户不喜欢吃香菜, 用户的项目代号是超体] 这个提取步骤是异步执行的不阻塞主响应流程。实测下来模型判断的准确率还不错偶尔会有误判但可以通过定期人工review来修正。4. 实操过程中踩过的坑与排查技巧4.1 流式输出中断问题最开始做流式输出的时候遇到一个诡异的问题本地测试一切正常部署到服务器后流式响应经常在中途断掉。排查了半天发现是 Nginx 的缓冲机制在作怪。Nginx 默认会缓冲后端返回的数据等缓冲区满了才发给客户端。对于流式响应这会导致客户端迟迟收不到数据或者收到一大块而不是逐字输出。解决办法是在 Nginx 配置里关掉缓冲location /api/chat/stream { proxy_pass http://127.0.0.1:8000; proxy_buffering off; proxy_cache off; proxy_set_header X-Accel-Buffering no; proxy_read_timeout 300s; }proxy_read_timeout也要调大因为模型生成慢的时候默认 60 秒可能不够。注意如果你用的是云服务商的负载均衡也要检查它们的缓冲设置。有些云厂商的 LB 默认开启缓冲需要在控制台手动关闭。4.2 模型调用超时与重试DeepSeek 的 API 偶尔会有响应慢的情况尤其是高峰期。如果不做超时和重试用户体验会很差。我的做法是设置合理的超时时间配合指数退避重试import asyncio from tenacity import retry, stop_after_attempt, wait_exponential retry( stopstop_after_attempt(3), waitwait_exponential(multiplier1, min2, max10) ) async def call_llm_with_retry(messages): try: return await asyncio.wait_for( client.chat.completions.create( modeldeepseek-chat, messagesmessages, timeout60 ), timeout90 ) except asyncio.TimeoutError: raise Exception(模型调用超时)这里用了两层超时asyncio.wait_for是外层总超时timeout参数是 HTTP 请求超时。为什么要两层因为有时候 HTTP 连接建立了但服务端不返回数据这种情况timeout参数不一定能覆盖需要外层兜底。4.3 记忆检索的相关性调优长期记忆检索最开始效果不好经常检索出一些不相关的内容。排查后发现两个问题问题一向量化模型的选择。Chroma 默认用的 embedding 模型对中文支持一般。我换成了专门的中文 embedding 模型后检索准确率明显提升。问题二检索策略太单一。纯语义检索对于「用户叫什么名字」这类精确查询效果不好。我加了一个关键词匹配的兜底逻辑如果语义检索的相似度都低于阈值就用关键词做一次精确匹配。async def retrieve_with_fallback(self, user_id: str, query: str, top_k: int 5): # 先做语义检索 results await self.semantic_search(user_id, query, top_k) # 检查相似度 if results and results[0][score] 0.7: return [r[content] for r in results] # 相似度不够补充关键词检索 keyword_results await self.keyword_search(user_id, query, top_k3) # 合并去重 seen set() merged [] for r in results keyword_results: if r[content] not in seen: seen.add(r[content]) merged.append(r[content]) return merged[:top_k]4.4 常见问题速查表问题现象可能原因排查方向解决方案流式输出卡顿Nginx 缓冲检查响应头关闭 proxy_buffering模型调用超时网络或服务端慢看日志时间戳加重试和超时记忆检索不准embedding 模型差测试相似度换中文模型上下文超长历史未压缩统计 token 数摘要压缩旧对话并发上不去同步阻塞检查 async 函数用 run_in_executor内存泄漏会话未清理监控内存加 TTL 清理这张表是我实际遇到问题后整理的基本覆盖了 80% 的常见故障。建议收藏出问题的时候按表排查能省不少时间。4.5 并发处理的注意事项智能体系统天然要面对并发问题。多个用户同时对话每个对话又涉及多次模型调用和数据库操作如果处理不好很容易出现性能瓶颈。我的经验是把耗时的操作异步化把共享的资源隔离好。具体来说模型调用全部用async不要用同步的requests库数据库连接用连接池不要每次请求都新建连接短期记忆用 Redis 而不是进程内存这样多 worker 之间能共享长期记忆的写入用队列异步处理不要阻塞主流程如果并发量真的很大可以考虑把模型调用单独拆成一个服务用消息队列解耦。不过对于个人智能体这个场景单机部署配合合理的异步设计扛住几十个并发没问题。5. 这套架构后续可以怎么扩展写到这里超体架构的核心部分基本讲完了。但一个真正好用的个人智能体还有很多可以打磨的地方。工具调用是下一步的重点。现在的智能体只能聊天如果能接入日历、笔记、邮件这些工具才能真正「下地干活」。我打算用 function calling 的方式实现工具注册和调用让模型自己决定什么时候该用什么工具。多模态输入也值得做。现在只能处理文字如果支持语音输入和图片理解使用场景会宽很多。语音可以用 Whisper 做转录图片可以用多模态模型理解。主动推送是个有意思的方向。现在的智能体是被动的你问它才答。如果它能根据你的日程和习惯主动提醒你「该开会了」「你关注的项目的更新了」那才真正像个「分身」。本地部署也在我的计划里。虽然 DeepSeek 的 API 已经很便宜了但有些敏感数据还是不想出本地。等手头的硬件到位我打算试试本地跑一个量化模型配合这套架构做完全离线的版本。这套架构我还在持续迭代后面会陆续把工具调用、多模态、主动推送这些模块的实现细节写出来。如果你也在做类似的东西欢迎交流踩坑经验。我个人在实际操作中的体会是不要追求一步到位先把最小闭环跑通再逐步加功能。我见过太多人一上来就想做个全能助手结果卡在架构设计上半年都没跑起来一个能用的版本。先让它能对话、能记住你然后再慢慢扩展这条路走起来踏实得多。