从零开始搭建一个简单的 AgentChat

项目地址 ：github项目地址
后端：python、LangChain、FastAPI
前端：Vue3
页面仿照 OpenClaw 部署页面 编写

项目前端页面展示

---

整篇文章结合项目内容，由AI简单梳理编写。

如果从零开始设计一个简单但能工作的 AgentChat，后端应该怎么拆，核心链路应该怎么走。

我会结合当前项目的实现，按模块把思路重新讲一遍。你可以把它理解成一版“重做架构时会怎么设计”的博客稿。

---

先想清楚: AgentChat 到底要解决什么问题

很多聊天项目一开始只做了“用户发一句，模型回一句”。这当然能跑，但很快就会遇到几个问题:

1. 模型不知道自己是谁。
2. 模型看不到历史上下文。
3. 模型不会查资料，也不会调用工具。
4. 模型拿不到长期记忆。
5. 前端只能等最终答案，过程不可见。

所以，一个最小可用的 AgentChat，至少应该有下面几层:

1. 请求入口层: 接收用户消息，负责流式输出。
2. Agent 组装层: 把提示词、历史、工具、技能拼成一个可运行 Agent。
3. 工具层: 给模型外挂能力。
4. 记忆层: 保存对话，并支持“回头翻旧账”。
5. 知识库层: 让模型能按主题检索资料。
6. 技能层: 根据用户意图，动态注入专门的规则。

这也是当前项目的后端主结构。

---

项目的核心调用链路

先看主流程。如果把这个系统抽象成一句话，大概就是:

WebSocket 收到消息 -> Agent 读取提示词和历史 -> 决定是否调用工具 -> 逐步产出结果 -> 写回记忆。

对应到项目里的后端文件，大致是:

1. 应用启动模块: 初始化数据库、向量模型、知识库、技能目录、MCP。
2. 对话入口模块: 负责流式事件编排。
3. Agent 核心模块: 组装真正运行的 Agent。
4. 工具注册模块: 注册系统可用工具。
5. 对话记忆模块: 负责历史保存和历史检索。
6. 知识库模块: 负责切片、入库、检索。
7. 技能模块: 负责技能发现、筛选、注入。

如果你要自己从零搭，最值得优先做的其实不是“多模型支持”，而是先把这条链路打通。

---

模块一: 请求入口层，只做编排，不做决策

入口层的职责很简单:

1. 接收用户输入。
2. 交给 Agent。
3. 把 Agent 的中间过程和最终结果流式发回去。
4. 在合适的时候落库。

当前项目里，这部分就是“对话入口层”。

最关键的函数是 stream_agent_events。它做了三件很重要的事。

1. 用队列把同步 Agent 包成异步流

async def stream_agent_events(input_text: str, session_id: str):
    history = DbChatMessageHistory(session_id=session_id)
    loop = asyncio.get_running_loop()
    queue: asyncio.Queue[dict] = asyncio.Queue()
    full_response = ""

这里的设计很实用。因为 Agent 执行本身可能是阻塞式的，但 WebSocket 希望异步流式发送，所以中间加了一层 asyncio.Queue 做桥接。

这类设计在 Agent 项目里很常见，原因很简单:

• 大模型调用是耗时操作。
• 工具调用也可能阻塞。
• 前端又希望边生成边显示。

所以入口层最适合做“事件搬运工”。

2. 区分三类输出: 文本块、工具动作、结束信号

for chunk in llm_client.think(input_text, session_id):
    if "output" in chunk:
        loop.call_soon_threadsafe(queue.put_nowait, {"type": "chunk", "content": output})
    elif "actions" in chunk:
        ...

这个拆分很关键。一个能用的 AgentChat，不应该只给前端最终答案，还应该给前端“过程感”。

所以这里把输出拆成:

1. chunk: 模型正在生成的正文。
2. action: 模型决定调用了哪个工具。
3. assistant_note: 工具调用前的说明性内容。
4. done: 本轮结束。

这会让用户明显感觉到: 这不是一个纯补全文本框，而是一个“会思考、会行动”的系统。

3. 最终答案才进入正式记忆

if event_type == "done":
    if full_response:
        history.add_message(AIMessage(content=full_response))

这个细节值得专门说一下。

中间的 action 和 assistant_note 会被当成系统消息单独记录，但真正作为“回答结果”写入长期聊天记录的，是完整的 AIMessage。这样做的好处是:

• 对话主历史更干净。
• 历史检索时更聚焦。
• 前端展示和后台记忆可以分层。

---

模块二: Agent 组装层，决定系统到底怎么“思考”

真正的核心，是那层负责组装 Agent 的运行逻辑。

如果只看架构职责，这个文件回答的是三个问题:

1. 用哪个模型。
2. 给模型什么上下文。
3. 让模型拥有哪些工具。

1. 先创建统一的 LLM 适配层

AgentLLM.__init__() 做的事并不复杂，但很重要: 把 OpenAI 和 Ollama 的初始化差异收敛到一处。

if provider == "ollama":
    self.llm = ChatOpenAI(
        api_key=OLLAMA_API_KEY,
        base_url=f"{OLLAMA_BASE_URL}/v1",
        model=model_name,
        timeout=timeout,
        temperature=API_TEMPERATURE,
    )
    return

这类设计的价值在于:

• 对上层透明。
• 后续切模型时不用改业务逻辑。
• Agent 逻辑和模型接入逻辑分离。

很多项目一开始把模型初始化写死在业务代码里，后面一换供应商就要到处改。这里提前抽出来，是对的。

2. 控制短期上下文窗口

think() 里有这样一段:

max_turns = 5
max_messages = max_turns * 2
all_history = [message for message in history.messages if message.type in {"human", "ai"}]
history_messages = all_history[-max_messages:] if len(all_history) > max_messages else all_history

这段代码体现了一个很实际的思想:

不是所有历史都该直接塞给模型。

短期上下文最有价值的是最近几轮对话，所以这里只取最近 5 轮。这样可以控制 token，也避免模型被陈旧信息干扰。

那更早的历史怎么办? 这个项目没有粗暴丢掉，而是交给了后面的“历史检索工具”。

也就是说，这里实际上做了一个经典的双层记忆设计:

1. 最近几轮 -> 直接放进 prompt。
2. 更早的内容 -> 需要时通过工具按语义检索。

这是一个非常适合 AgentChat 的结构。

3. 系统提示词不是一份，而是“拼接出来的”

with open("definition/IDENTITY.md", "r", encoding="utf-8") as file:
    identity_prompt = file.read()
skills_catalog_prompt = render_skill_catalog_prompt()
selected_skills_prompt = render_relevant_skills_prompt(input_text)
system_prompt = identity_prompt + "\n" + tool_registry.getAvailableTools()

这段代码特别值得拿出来讲，因为它已经不是“固定 prompt”思路了，而是“可组装 prompt”思路。

系统提示词由四部分组成:

1. IDENTITY.md: 定义 Agent 的身份、说话方式、行为边界。
2. 工具描述: 告诉模型自己有什么能力。
3. 技能目录: 告诉模型系统里有哪些可选技能。
4. 命中的技能全文: 只把当前请求真正相关的技能注入进去。

如果从零搭项目，我会把这一步理解为:

我们不是在给模型一段大而全的总规则，而是在按场景动态组装工作说明书。

4. 用 LangChain 的 tools agent 作为执行内核

prompt = ChatPromptTemplate.from_messages(
    [
        SystemMessagePromptTemplate.from_template(system_prompt),
        MessagesPlaceholder("chat_history"),
        HumanMessagePromptTemplate.from_template("{input}"),
        MessagesPlaceholder("agent_scratchpad"),
    ]
)

agent = create_openai_tools_agent(self.llm, tool_registry.getToolsList(), prompt)

这个结构非常标准，但也非常好用:

1. system_prompt: 角色和规则。
2. chat_history: 最近几轮上下文。
3. input: 当前用户问题。
4. agent_scratchpad: Agent 的中间推理和工具轨迹。

本质上，这里是在把“普通聊天”升级为“可调用工具的推理型聊天”。

5. 用上下文变量把 session_id 传给工具

token = set_current_session_id(session_id)
stream = agent_executor.stream({"input": input_text, "chat_history": history_messages})

然后在结束时:

reset_current_session_id(token)

这一步看起来小，但设计非常聪明。因为有些工具本身需要知道“当前是哪一个会话”，比如“检索更早历史”这个工具，它就需要按当前会话检索旧消息。

如果把 session_id 层层当参数传下去，会让工具接口很难看。这里用 ContextVar 存当前会话，就能让工具在不污染参数签名的前提下拿到上下文。

这就是“会话上下文模块”的价值。

---

模块三: 工具系统，先做“注册中心”，再做“具体能力”

Agent 和普通聊天最大的区别，就是它不只会回答，还会调用外部能力。

当前项目的工具体系拆成了两层:

1. 工具注册中心: 管理工具注册。
2. 工具装配层: 真正把工具组装起来。

1. 工具注册中心为什么要单独存在

ToolExecutor 很轻量，但它做对了一件事: 不把工具列表散落在 Agent 主代码里。

class ToolExecutor:
    def __init__(self):
        self.tools: Dict[str, Dict[str, Any]] = {}

    def registerTool(self, name: str, description: str, func: callable):
        self.tools[name] = {"description": description, "func": func}

这层抽象带来的好处是:

• Agent 只关心“我有哪些工具”，不关心工具从哪来。
• MCP 工具、知识库工具、系统工具可以统一注册。
• 以后做工具开关、权限控制、分组展示会更容易。

2. 真正重要的是工具描述文本

来看工具装配这一层:

tool_executor.registerTool(
    "get_knowledge_definitions",
    "列出所有知识库定义及其 source_key 和描述。在调用 retrieve_profile 之前应先调用此工具，以便选择最相关的知识库。",
    get_knowledge_definitions,
)

这段代码里最重要的不是函数名，而是描述。

因为对大模型来说，工具描述本身就是“使用说明”。比如这里明确写了:

调用 retrieve_profile 前，应先调用 get_knowledge_definitions。

这等于把工具调用顺序也教给了模型。

很多 Agent 项目工具明明都写了，但模型不会用，往往不是函数有问题，而是说明文字太弱。

3. 这个项目里最实用的几类工具

从架构角度看，当前工具可以分成四组:

1. 信息检索: web_search、hot_search、get_weather、get_current_time
2. 记忆检索: search_early_history
3. 知识库检索: get_knowledge_definitions、retrieve_profile
4. 外部扩展: safe_shell 和 MCP 工具

这已经覆盖了一个最小 AgentChat 最常见的能力面。

如果你自己从零做，我建议第一版先只保留三种:

1. 时间工具
2. Web 搜索
3. 历史检索

先让模型学会“什么时候该查，什么时候该答”，系统就活了。

---

模块四: 记忆系统，不是只存历史，而是要“能找回来”

很多人做聊天记忆时，只想到把消息存到数据库里。但 AgentChat 真正需要的是:

不仅能存，还要能按语义找回来。

这部分主要由“对话记忆模块”承担。

1. 每条消息入库时顺手生成 embedding

msg_dict = message_to_dict(message)
doc_content = json.dumps(msg_dict, ensure_ascii=False)
embedding_vector = embeddings.embed_query(doc_content)
embedding_str = json.dumps(embedding_vector)

这一步非常关键。因为一旦消息入库时就把向量算好，后面做历史语义检索就很直接。

数据库里这几个字段基本就够用了:

1. session_id: 属于哪个会话。
2. message_type: 人类、AI、系统消息。
3. content: 原始消息内容。
4. embedding: 向量。
5. timestamp: 时间戳。

对应的数据表就是聊天消息表 ChatMessage。

2. 短期缓存放 Redis，正式数据放 PostgreSQL

cached_records = get_cached_history_messages(self.session_id)
if cached_records:
    return self._records_to_messages(cached_records)

这说明当前实现不是每次都直接打数据库，而是优先读 Redis 缓存，拿不到再查 PostgreSQL。

这个组合非常适合聊天系统:

• Redis 负责高频读取。
• PostgreSQL 负责持久化。

如果从零搭，哪怕暂时不接 Redis，也建议把缓存层接口留出来，因为对话历史一定是高频访问数据。

3. 历史检索只找“更早的内容”，避免和短期上下文打架

search_early_history() 的逻辑很值得借鉴:

skip_messages = recent_turns_to_skip * 2
candidate_results = results[:-skip_messages] if len(results) > skip_messages else []

这段代码的意思是:

最近几轮已经直接塞进 prompt 了，所以语义检索时主动跳过它们，只在更老的消息里找。

这个设计很聪明，因为它避免了两个问题:

1. 重复把最近内容再查一遍。
2. 检索结果总被最近消息“霸榜”。

这就是为什么“历史检索工具”虽然很简单，但很有价值。它不是“查全部历史”，而是“查前文里被遗忘但可能还有用的部分”。

---

模块五: 知识库系统，让 Agent 会“按主题查资料”

聊天历史解决的是“这个用户以前说过什么”，知识库解决的是“系统额外知道什么”。

这部分就是知识库模块在做的事情。

1. 知识库的最小设计

从当前实现里，可以总结出一套很清晰的知识库模型:

1. 原始文本来源
2. 文本切片
3. 每个切片的 embedding
4. 一个独立的 definition 表，记录每个知识源的摘要信息

其中 KnowledgeChunk 存内容切片，KnowledgeDefinition 存来源定义。这两个表分开是对的，因为:

• 切片表适合检索。
• 定义表适合给模型先“选库”。

2. 先切片，再向量化

splitter = RecursiveCharacterTextSplitter(
    chunk_size=DEFAULT_CHUNK_SIZE,
    chunk_overlap=DEFAULT_CHUNK_OVERLAP,
    separators=["\n## ", "\n# ", "\n\n", "\n", " "],
)

这一步很典型。知识库不能把整篇文档直接喂给模型，一般都要先拆块。这里用了递归切分器，而且优先按标题、段落、换行来切，比纯字符硬切更自然。

3. 检索之前，先让模型选对知识源

当前项目没有让模型直接全库搜索，而是设计了两步式调用:

1. get_knowledge_definitions
2. retrieve_profile

这个顺序非常值得讲。因为它不是让模型盲搜，而是:

1. 先看有哪些知识源可选。
2. 选一个最相关的 source_key。
3. 再去这个知识源下做向量检索。

这种设计有三个好处:

1. 降低误检索。
2. 工具调用路径更清晰。
3. 对模型更友好，减少“搜错库”的概率。

负责执行知识检索的那个工具函数本身很短，但它背后连接的是完整的知识库查询链路。

4. 默认知识库不只存业务资料，也存 Agent 设定

项目启动时，会先做这两步:

ensure_knowledge_definitions()
ensure_default_definition_sources()

而默认定义源大致是这样组织的:

DEFINITION_SOURCES = {
    "SOUL": Path("definition/SOUL.md"),
    "USER": Path("definition/USER.md"),
}

这说明这里的“知识库”并不只是外挂文档，还承担了两类长期资料:

1. Agent 自己的人设与行为信息。
2. 当前用户的资料与偏好。

这是一种很实用的设计。它把“设定”和“知识”统一成可检索的数据源，而不是全部硬塞进 system prompt。

---

模块六: 技能系统，让 Prompt 不再是固定大杂烩

如果说工具解决的是“能做什么”，那技能解决的是“这次该按什么规则做”。

这部分主要由技能模块负责。

1. 技能的本质是什么

在这个项目里，一个技能本质上就是一个带说明的目录，核心文件是 SKILL.md。

系统会先扫描 skills/ 目录，找到所有已安装技能:

for child in sorted(skills_dir.iterdir(), key=lambda item: item.name.lower()):
    skill_file = child / "SKILL.md"

然后提取每个技能的标题和描述，形成“技能目录”。

2. 技能不是全量注入，而是按请求动态筛选

matched_skills = select_relevant_skills(user_input=user_input, max_skills=max_skills)

这个思路非常重要。因为如果把所有技能全文都塞进 prompt，问题会很快出现:

1. token 膨胀
2. 指令互相干扰
3. 模型抓不到重点

所以这里做了两层筛选:

1. 词法相关性
2. embedding 语义相关性

最后只选少数最相关技能注入 prompt。

这其实就是把 RAG 思想用在了 Prompt 指令层。

3. 为什么技能目录和技能全文要分开

render_skill_catalog_prompt() 和 render_relevant_skills_prompt() 是分开的。

这背后的设计逻辑是:

1. 先告诉模型“系统里有哪些技能可选”。
2. 再把本次最相关的技能全文塞进去。

这样模型既有全局认知，又不会被无关说明淹没。

如果从零实现 AgentChat，这一层不是必需的，但一旦系统开始长大，它会非常有用。因为它解决的是 Prompt 工程里最容易失控的问题: 规则越来越多，但每次只需要其中一小部分。

---

模块七: 数据模型，决定后面扩展难不难

一个聊天 Agent 的后端数据表不用太多，但最好一开始就把边界分清楚。

当前项目的核心表有三张:

1. ChatMessage: 会话消息
2. KnowledgeChunk: 知识切片
3. KnowledgeDefinition: 知识源定义

这三张表刚好对应三类数据:

1. 对话过程数据
2. 可检索文本数据
3. 检索数据的目录信息

这个拆法的好处是:

• 会话和知识分离
• 检索和展示分离
• 以后加权限、标签、来源管理都比较顺手

如果一开始就把所有东西都塞进一张“documents”表，短期看省事，后面会越来越难维护。

---

模块八: 启动初始化，确保系统一启动就“可用”

在启动阶段，应用会初始化这些资源:

init_db()
init_embeddings()
get_redis_client().ping()
ensure_knowledge_definitions()
ensure_default_definition_sources()
ensure_skills_dir()
mcp_manager.start()

这段代码其实体现了一个很成熟的后端思路:

不要等第一条请求来了再懒加载所有东西，而是把关键依赖在启动阶段准备好。

尤其是下面几项，最好提前完成:

1. 数据库表初始化
2. embedding 模型预热
3. 默认知识源同步
4. 技能目录准备

这样用户第一次发消息时，系统就不会因为“现场加载模型”而卡住很久。