企业官网建设流程全解析

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_key="your-api-key", base_url="https://api.deepseek.com/v1" ) async def chat(messages: list[dict]) -> str: response = await client.chat.completions.create( model="deepseek-chat", messages=messages, temperature=0.7, max_tokens=2048 ) return response.choices[0].message.content

这段代码就是最基础的调用封装。但注意，直接这样用是不够的，因为每次调用都是无状态的，模型不记得上一轮说了什么。要让它「记得住」，就得在messages里把历史对话带上。而历史对话怎么存、存多少、怎么检索，就是认知层要解决的问题。

2.3 记忆机制：让智能体真正「懂你」的关键

这是整套架构里我最花心思的部分。市面上大部分教程讲智能体，记忆这块要么一笔带过，要么就是简单地把对话历史塞进 context。但真正的「越用越懂你」，需要的是分层记忆。

我把记忆分成三类：

短期记忆好理解，就是最近几轮对话。但这里有个坑：不能无限制地往 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

这里解释几个关键依赖的选择理由：

• uvicorn：ASGI 服务器，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_id=request.user_id, message=request.message, session_id=request.session_id ): yield f"data: {chunk}\n\n" return StreamingResponse( event_generator(), media_type="text/event-stream" )

这里用的是 SSE（Server-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_length=1, max_length=4000)

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_id=user_id, query=message, top_k=5 ) # 2. 获取短期记忆 short_term = await self.memory.get_short_term(session_id) # 3. 组装 prompt messages = self._build_messages( long_term=long_term, short_term=short_term, user_message=message ) # 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_id=user_id, conversation=f"用户: {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_context=memory_text)} ] messages.extend(short_term) messages.append({"role": "user", "content": user_message}) return messages

3.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(hours=ttl_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( path=path, settings=ChromaSettings(anonymized_telemetry=False) ) self.collection = self.client.get_or_create_collection( name="user_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_results=top_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( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10) ) async def call_llm_with_retry(messages): try: return await asyncio.wait_for( client.chat.completions.create( model="deepseek-chat", messages=messages, timeout=60 ), timeout=90 ) 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_k=3) # 合并去重 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 常见问题速查表

这张表是我实际遇到问题后整理的，基本覆盖了 80% 的常见故障。建议收藏，出问题的时候按表排查，能省不少时间。

4.5 并发处理的注意事项

智能体系统天然要面对并发问题。多个用户同时对话，每个对话又涉及多次模型调用和数据库操作，如果处理不好，很容易出现性能瓶颈。

我的经验是：把耗时的操作异步化，把共享的资源隔离好。具体来说：

• 模型调用全部用async，不要用同步的requests库
• 数据库连接用连接池，不要每次请求都新建连接
• 短期记忆用 Redis 而不是进程内存，这样多 worker 之间能共享
• 长期记忆的写入用队列异步处理，不要阻塞主流程

如果并发量真的很大，可以考虑把模型调用单独拆成一个服务，用消息队列解耦。不过对于个人智能体这个场景，单机部署配合合理的异步设计，扛住几十个并发没问题。

5. 这套架构后续可以怎么扩展

写到这里，超体架构的核心部分基本讲完了。但一个真正好用的个人智能体，还有很多可以打磨的地方。

工具调用是下一步的重点。现在的智能体只能聊天，如果能接入日历、笔记、邮件这些工具，才能真正「下地干活」。我打算用 function calling 的方式实现工具注册和调用，让模型自己决定什么时候该用什么工具。

多模态输入也值得做。现在只能处理文字，如果支持语音输入和图片理解，使用场景会宽很多。语音可以用 Whisper 做转录，图片可以用多模态模型理解。

主动推送是个有意思的方向。现在的智能体是被动的，你问它才答。如果它能根据你的日程和习惯，主动提醒你「该开会了」「你关注的项目的更新了」，那才真正像个「分身」。

本地部署也在我的计划里。虽然 DeepSeek 的 API 已经很便宜了，但有些敏感数据还是不想出本地。等手头的硬件到位，我打算试试本地跑一个量化模型，配合这套架构做完全离线的版本。

这套架构我还在持续迭代，后面会陆续把工具调用、多模态、主动推送这些模块的实现细节写出来。如果你也在做类似的东西，欢迎交流踩坑经验。我个人在实际操作中的体会是：不要追求一步到位，先把最小闭环跑通，再逐步加功能。我见过太多人一上来就想做个全能助手，结果卡在架构设计上，半年都没跑起来一个能用的版本。先让它能对话、能记住你，然后再慢慢扩展，这条路走起来踏实得多。