1. 从标题拆解一个Agent系统的四根支柱
第一次看到“Hermes Agent能力篇:会话、工具、MCP、记忆”这个标题,我脑子里冒出来的第一个判断是:这是一篇讲Agent运行时骨架的内容,而不是讲某个具体业务场景的落地案例。会话、工具、MCP、记忆这四个词放在一起,基本覆盖了一个智能体从“能对话”到“能干活”再到“能记住”的完整链路。如果你正在做Agent相关的开发,或者准备把大模型能力接入到自己的业务系统里,这四个模块就是你绕不开的核心。
先说清楚这个标题对应的内容到底在讲什么。Hermes在这里可以理解为一个Agent框架或者Agent运行时的代称,它要解决的问题是:如何让一个大模型驱动的智能体在多轮交互中保持上下文、如何让它调用外部能力、如何通过标准化协议接入更多工具、以及如何让它在跨会话场景下记住关键信息。这四个能力分别对应四个技术模块:会话管理负责单次交互内的上下文维护,工具调用负责让模型能执行动作,MCP负责工具接入的标准化,记忆系统负责跨会话的信息持久化。
适合谁来参考?如果你只是用现成的对话产品,这篇内容可能偏底层;但如果你在自研Agent、做企业级智能助手、或者想把LLM接入到已有的工作流系统里,那这四个模块的设计思路和踩坑经验就非常值得花时间看。我自己的经验是,很多团队在Demo阶段只关注“模型能不能答对”,一到生产环境就发现会话串了、工具调不通、记忆丢了,问题全出在这四个模块的工程实现上。
接下来我会按“整体设计思路—核心细节—实操实现—问题排查”的顺序,把这四个能力逐个拆开讲,中间会穿插我在实际项目里踩过的坑和验证过的方案。内容偏工程实践,不堆概念,能直接抄的部分我会尽量给到具体参数和代码片段。
2. 整体架构设计与模块拆解思路
2.1 为什么是这四个模块而不是别的
一个Agent系统要跑起来,最朴素的形态就是“用户输入—模型推理—返回结果”。但这个形态只能做问答,做不了复杂任务。要让它能干活,至少需要三样东西:记住当前对话说了什么(会话)、能调用外部能力(工具)、下次还能记得上次的事(记忆)。而MCP是在工具数量变多之后,为了解决“每个工具都要单独写适配层”这个问题引入的标准化协议。
我见过一些团队一开始只做会话和工具,记忆用简单的数据库存一下,结果工具接入到十几个之后,维护成本急剧上升,每接一个新工具就要改一遍调度逻辑。MCP的价值就在这里:它把工具的发现、描述、调用统一成一套协议,Agent侧只需要实现一次MCP客户端,后面接多少工具都是配置问题。
所以这四个模块的关系不是并列的,而是有层次的。会话是基础层,工具是能力层,MCP是接入层,记忆是持久层。设计的时候如果把它们混在一起做,后期扩展会很痛苦。我的建议是在代码结构上就把这四个模块分开,各自有独立的接口和存储,通过一个调度器串联。
2.2 模块之间的数据流怎么走
理解数据流是做好设计的前提。一次典型的Agent调用,数据流向大致是这样的:用户输入进入会话模块,会话模块从存储中取出当前会话的历史消息,拼装成模型需要的上下文格式;模型推理后如果决定调用工具,工具模块根据MCP协议描述的工具列表执行调用,拿到结果后回填到会话上下文;模型继续推理直到给出最终回复;回复返回给用户的同时,会话模块把本轮消息写入存储;如果本轮涉及需要长期保留的信息,记忆模块会抽取关键内容写入长期存储。
这个流程里有两个容易出问题的地方。一是会话上下文的拼装策略,历史消息不能无限往里塞,要有截断或摘要机制;二是工具调用结果的回填格式,如果格式和模型预期的不一致,模型会反复调用同一个工具或者直接忽略结果。这两个点后面会详细讲。
2.3 选型时的几个关键取舍
会话存储用什么?内存、Redis、还是关系型数据库?我的经验是看会话生命周期。如果只是单次会话内的上下文,内存加定期清理就够了;如果需要支持会话恢复(比如用户刷新页面后继续),那就要用Redis这类带过期时间的存储;如果还要做会话审计和分析,那就得上关系型数据库。很多团队一上来就用最重的方案,结果发现90%的会话根本不需要持久化,白白增加了延迟。
工具调用的执行方式也有取舍。同步调用简单直接,但一个工具卡住整个Agent就卡住;异步调用灵活,但需要处理回调、超时、重试这些复杂逻辑。我一般建议对响应时间敏感的工具用同步,对耗时长的工具用异步加轮询。MCP协议本身是支持这两种模式的,具体用哪种要看工具的实际特性。
记忆系统的设计取舍更微妙。全量存储成本高但信息完整,摘要存储成本低但会丢细节。我试过纯摘要的方案,结果用户问“上次我们讨论的那个方案里提到的第三个参数是多少”,摘要里根本没这个信息。后来改成“摘要加关键实体抽取”的混合方案,摘要保留对话脉络,实体抽取保留具体数值和名称,效果好了很多。
3. 会话管理的核心细节与实操要点
3.1 会话上下文的拼装策略
会话管理的核心任务是把历史消息拼装成模型能理解的上下文。听起来简单,但实际做的时候有几个坑。第一个坑是消息顺序。有些模型对消息顺序敏感,system消息必须在最前面,user和assistant消息要严格交替。我见过因为消息顺序错乱导致模型完全忽略system指令的情况,排查了半天才发现是存储层返回的消息没有按时间排序。
第二个坑是上下文长度控制。模型的上下文窗口是有限的,历史消息不能无限往里塞。常见的策略有三种:滑动窗口、摘要压缩、关键消息保留。滑动窗口最简单,只保留最近N轮对话,但会丢失早期的重要信息;摘要压缩是把早期对话总结成一段话,保留脉络但丢细节;关键消息保留是标记重要消息(比如包含用户明确需求的),优先保留这些。我一般用滑动窗口加关键消息保留的组合,窗口大小根据模型上下文窗口的70%来定,留30%给当前轮的工具调用结果和模型输出。
具体参数上,假设模型上下文窗口是128K token,我会把历史消息控制在80K以内,当前轮的工具调用结果预留20K,模型输出预留10K,剩下18K作为缓冲。这个比例不是固定的,如果工具返回结果特别大,就要相应压缩历史消息。
3.2 会话状态的存储与恢复
会话状态不只是消息列表,还包括一些元信息:会话创建时间、最后活跃时间、当前使用的模型、已调用的工具列表等。这些信息在会话恢复时都要用到。存储结构上,我习惯用两层:一层是会话元信息,用Redis的Hash结构存,key是会话ID,field是各个元信息字段;另一层是消息列表,用Redis的List存,每个元素是一条消息的JSON。
恢复的时候先读元信息,再读消息列表,然后按时间排序拼装。这里有个细节:消息的ID生成要用单调递增的方式,不能用UUID,否则排序会乱。我一般用时间戳加序列号的方式生成消息ID,保证同一毫秒内的消息也能正确排序。
注意:会话恢复时不要直接信任存储里的消息顺序,一定要按消息ID重新排序。我踩过一次坑,Redis的List在并发写入时顺序会乱,导致恢复出来的对话上下文错位,模型完全无法理解。
3.3 多会话并发时的隔离问题
一个Agent服务通常要同时处理多个用户的会话,隔离做不好就会出现A用户的上下文串到B用户的情况。隔离的关键是会话ID的生成和传递。会话ID必须在用户第一次请求时生成,之后每次请求都带上,服务端根据会话ID去取对应的上下文。我见过有的实现用用户ID当会话ID,结果同一用户开两个窗口对话就串了。
并发写入是另一个问题。同一会话内如果有多条消息同时写入,不加锁的话顺序会乱。我的做法是在会话级别加一个分布式锁,写入前先获取锁,写完释放。锁的粒度要细,只锁当前会话,不要锁整个存储。Redis的Redlock或者基于SETNX的简单锁都能用,关键是设置合理的超时时间,避免死锁。
4. 工具调用与MCP协议落地细节
4.1 工具描述的结构化设计
工具调用的第一步是让模型知道有哪些工具可用。工具描述的质量直接决定模型能不能正确选择工具。一个完整的工具描述至少包含:工具名称、功能说明、参数列表(每个参数的类型、是否必填、说明)、返回值说明。我见过很多实现只给工具名称和一句话说明,结果模型要么不调用,要么调用时参数传错。
参数说明要具体。比如一个“查询天气”的工具,参数“城市”要说明是城市名称还是城市代码,是中文还是英文。我一般会在参数说明里给示例值,模型看到示例后传参的准确率会明显提升。返回值说明也要写清楚,特别是返回结构复杂的时候,要说明每个字段的含义,否则模型拿到结果也不知道怎么用。
4.2 MCP协议的核心机制
MCP的核心是把工具的发现和调用标准化。Agent侧实现一个MCP客户端,工具侧实现一个MCP服务端,两者通过标准协议通信。协议主要包含几个部分:工具列表查询、工具调用、调用结果返回。工具列表查询让Agent知道当前有哪些工具可用,工具调用让Agent执行具体动作,结果返回把执行结果传回来。
MCP的好处是解耦。以前每接一个新工具,Agent侧就要改代码;现在只要工具侧实现了MCP服务端,Agent侧不用动。我实测下来,接一个新工具的时间从原来的半天缩短到半小时,主要时间花在写工具本身的逻辑上,协议适配的部分几乎为零。
MCP的通信方式支持多种传输层,常见的有标准输入输出和HTTP。标准输入输出适合本地工具,启动快、延迟低;HTTP适合远程工具,跨网络调用方便。选哪种看工具部署在哪里,本地工具用标准输入输出,远程工具用HTTP。
4.3 工具调用的错误处理与重试
工具调用失败是常态,网络抖动、服务超时、参数错误都会导致失败。错误处理做不好,Agent要么卡死,要么给用户返回一堆错误信息。我的做法是分三层处理:第一层是参数校验,调用前检查参数是否完整、类型是否正确,不合法直接返回错误让模型重新生成;第二层是调用超时,设置合理的超时时间(一般5到10秒),超时后返回超时错误;第三层是重试,对可重试的错误(如网络抖动)自动重试,重试次数一般2到3次,每次间隔递增。
重试要注意幂等性。查询类工具重试没问题,但写入类工具重试可能导致重复写入。我的做法是在工具描述里标记是否幂等,非幂等工具不自动重试,而是把错误返回给模型,让模型决定是否重新调用。
提示:工具调用的超时时间不要设得太短。我试过设2秒,结果很多正常的数据库查询都被判超时,模型反复重试反而更慢。后来改成8秒,超时率降到1%以下。
4.4 工具调用结果的格式化回填
工具调用结果回填给模型时,格式很关键。我见过直接把JSON字符串塞回去的,模型理解起来很吃力。更好的做法是把结果转成自然语言描述,或者至少是结构化的文本。比如查询天气返回{"temp": 25, "condition": "sunny"},回填时写成“当前温度25度,天气晴朗”,模型后续推理会顺畅很多。
如果结果数据量大,还要做截断或摘要。比如查询返回100条记录,全塞回去会占满上下文,模型也处理不过来。我一般只回填前10条加一个总数说明,模型需要更多数据时会再次调用工具并指定分页参数。
5. 记忆系统的设计与实现
5.1 短期记忆与长期记忆的分工
记忆系统要解决的是跨会话的信息保留问题。短期记忆就是当前会话的上下文,这个由会话模块管理;长期记忆是跨会话的,需要单独设计。长期记忆又分两种:一种是用户画像类的,比如用户的偏好、习惯,这种变化慢,可以定期更新;另一种是事实类的,比如用户上次提到的项目名称、关键数据,这种需要精确存储和检索。
我的做法是把长期记忆分成两个存储:画像存储用键值对,key是用户ID,value是画像JSON;事实存储用向量数据库,把事实文本向量化后存储,检索时用相似度匹配。画像存储简单直接,事实存储灵活但需要处理向量化质量和检索准确率的问题。
5.2 记忆的写入时机与抽取策略
记忆不是每轮对话都要写,那样存储成本高且噪音大。我的策略是:用户明确表达偏好时写入画像,用户提到具体事实时写入事实存储,其他情况不写。判断“明确表达”和“具体事实”可以用规则加模型判断的方式。规则部分用关键词匹配,比如“我喜欢”“我习惯”“我的项目是”这类句式;模型判断部分用一个轻量级的分类模型,判断当前轮是否包含值得记忆的信息。
抽取策略上,事实类记忆要抽取实体和关系。比如用户说“我们团队用的是Python 3.11,部署在K8s上”,抽取出的实体是“团队”“Python 3.11”“K8s”,关系是“使用”“部署在”。这些结构化信息存储后,检索时可以精确匹配,比纯文本相似度更准。
5.3 记忆检索的触发与排序
记忆检索的触发时机是在会话开始时和每轮对话前。会话开始时检索用户画像,把画像信息注入system消息;每轮对话前检索相关事实,把事实信息注入当前上下文。检索的触发条件要控制好,不是每轮都检索,而是当用户输入包含特定意图时检索。比如用户问“上次那个方案”,就触发事实检索;用户只是打招呼,就不检索。
检索结果的排序用相似度加时间衰减。相似度高的排前面,时间近的排前面。我一般用相似度占70%权重,时间衰减占30%。时间衰减用指数衰减,半衰期设30天,超过半年的记忆权重降到很低。
5.4 记忆的更新与冲突处理
记忆更新时可能遇到冲突。比如用户之前说“喜欢红色”,后来说“喜欢蓝色”,两条记忆冲突。我的处理方式是保留最新的一条,旧的那条标记为过期但不删除,用于审计。如果冲突的是事实类记忆,比如项目名称变了,同样保留最新,旧版本标记过期。
更新策略上,画像类记忆用覆盖更新,事实类记忆用追加更新。画像覆盖是因为画像代表当前状态,旧状态没有保留价值;事实追加是因为事实有历史价值,可能后续会用到。
6. 实操过程与核心环节实现
6.1 环境准备与依赖安装
先列一下我用的技术栈:Python 3.11、Redis 7.0、PostgreSQL 15、ChromaDB(向量存储)、FastAPI(服务框架)。这些版本是我实测下来比较稳定的组合,Python 3.11的性能比3.10好不少,Redis 7.0的RedisJSON功能对存储会话元信息很方便。
安装依赖用pip:
pip install fastapi uvicorn redis psycopg2-binary chromadb openai tiktokentiktoken是用来计算token数的,会话上下文控制需要精确的token计数。ChromaDB用来做向量存储,轻量且API简单,适合中小规模的事实记忆存储。
6.2 会话模块的代码实现
会话模块的核心是两个类:SessionManager和SessionStore。SessionManager负责业务逻辑,SessionStore负责存储读写。
import redis import json import time from typing import List, Dict class SessionStore: def __init__(self, redis_client): self.redis = redis_client def get_session_meta(self, session_id: str) -> Dict: meta = self.redis.hgetall(f"session:{session_id}:meta") return {k.decode(): v.decode() for k, v in meta.items()} def set_session_meta(self, session_id: str, meta: Dict): self.redis.hset(f"session:{session_id}:meta", mapping=meta) self.redis.expire(f"session:{session_id}:meta", 86400 * 7) def append_message(self, session_id: str, message: Dict): msg_id = f"{int(time.time() * 1000)}-{self.redis.incr('msg_seq')}" message["msg_id"] = msg_id self.redis.rpush(f"session:{session_id}:messages", json.dumps(message)) self.redis.expire(f"session:{session_id}:messages", 86400 * 7) def get_messages(self, session_id: str, limit: int = 100) -> List[Dict]: raw = self.redis.lrange(f"session:{session_id}:messages", -limit, -1) messages = [json.loads(m) for m in raw] messages.sort(key=lambda x: x["msg_id"]) return messages这段代码里有两个细节值得说。一是消息ID的生成用了时间戳加自增序列,保证单调递增;二是读取时按msg_id重新排序,防止Redis List顺序错乱。过期时间设了7天,根据业务需要可以调整。
6.3 工具调用与MCP客户端的实现
MCP客户端的核心是工具列表获取和工具调用两个方法。我用HTTP传输层举例:
import httpx class MCPClient: def __init__(self, server_url: str): self.server_url = server_url self.tools = [] async def list_tools(self) -> List[Dict]: async with httpx.AsyncClient() as client: resp = await client.get(f"{self.server_url}/tools") self.tools = resp.json()["tools"] return self.tools async def call_tool(self, tool_name: str, params: Dict, timeout: int = 8) -> Dict: async with httpx.AsyncClient(timeout=timeout) as client: try: resp = await client.post( f"{self.server_url}/tools/{tool_name}/call", json={"params": params} ) return resp.json() except httpx.TimeoutException: return {"error": "timeout", "retryable": True} except Exception as e: return {"error": str(e), "retryable": False}超时设8秒是经验值,前面提过。返回结果里带retryable标记,方便上层决定是否重试。
6.4 记忆模块的向量化与检索
记忆模块用ChromaDB做向量存储,写入和检索的代码大致如下:
import chromadb class MemoryStore: def __init__(self, persist_dir: str): self.client = chromadb.PersistentClient(path=persist_dir) self.collection = self.client.get_or_create_collection("facts") def add_fact(self, user_id: str, fact: str, metadata: Dict): self.collection.add( documents=[fact], metadatas=[{**metadata, "user_id": user_id}], ids=[f"{user_id}-{int(time.time() * 1000)}"] ) def search_facts(self, user_id: str, query: str, top_k: int = 5) -> List[str]: results = self.collection.query( query_texts=[query], n_results=top_k, where={"user_id": user_id} ) return results["documents"][0] if results["documents"] else []检索时用user_id过滤,保证只检索当前用户的记忆。top_k设5是平衡召回率和上下文长度的结果,设太大上下文会被记忆占满。
6.5 四个模块的串联调度
最后用一个Agent类把四个模块串起来:
class HermesAgent: def __init__(self, session_store, mcp_client, memory_store, llm_client): self.session_store = session_store self.mcp_client = mcp_client self.memory_store = memory_store self.llm = llm_client async def chat(self, session_id: str, user_input: str) -> str: # 1. 加载会话历史 messages = self.session_store.get_messages(session_id) # 2. 检索相关记忆 facts = self.memory_store.search_facts(session_id, user_input) # 3. 拼装上下文 context = self._build_context(messages, facts, user_input) # 4. 模型推理与工具调用循环 response = await self._reason_and_act(context) # 5. 写入会话 self.session_store.append_message(session_id, {"role": "user", "content": user_input}) self.session_store.append_message(session_id, {"role": "assistant", "content": response}) return response这个调度逻辑里,_reason_and_act是核心循环,模型返回工具调用请求就执行工具,执行完把结果回填继续推理,直到模型返回最终回复。循环次数要设上限,一般5到10次,防止模型陷入死循环。
7. 常见问题与排查技巧实录
7.1 会话串号问题排查
会话串号是最常见也最致命的问题。表现是A用户看到了B用户的对话内容。排查思路:先检查会话ID的生成和传递链路,确认每个请求都带了正确的会话ID;再检查存储层的key是否包含了用户ID,如果只用会话ID做key,不同用户的会话ID碰撞就会串号。我的做法是存储key用user:{user_id}:session:{session_id}的格式,双重隔离。
7.2 工具调用超时与重试失效
工具调用超时后重试仍然失败,常见原因是重试时没有重新建立连接。httpx的AsyncClient在超时后连接可能已经断开,重试时要新建客户端。另外重试间隔要递增,我一般用1秒、2秒、4秒的间隔,避免短时间内大量重试压垮工具服务。
7.3 记忆检索不准确
记忆检索不准确通常有两个原因:向量化质量差和检索参数不合理。向量化质量差可以用更好的embedding模型解决,我实测下来,专门的中文embedding模型比通用模型在中文事实检索上准确率高20%左右。检索参数方面,相似度阈值要设,低于阈值的不要返回,我一般设0.7,低于这个值的结果噪音太大。
7.4 上下文超长导致模型截断
上下文超长时模型会截断输入,表现是模型忽略了前面的指令。排查方法是打印实际发送给模型的token数,和模型上下文窗口对比。控制策略前面讲过,滑动窗口加关键消息保留。这里补充一个技巧:工具调用结果如果特别大,可以先摘要再回填,摘要用一个小模型做,成本低且效果好。
7.5 常见问题速查表
| 问题现象 | 可能原因 | 排查方法 | 解决方案 |
|---|---|---|---|
| 会话串号 | 会话ID碰撞或存储key设计不当 | 检查存储key格式 | key加用户ID前缀 |
| 工具调用超时 | 超时时间过短或工具服务慢 | 打印调用耗时 | 超时调到8秒,加重试 |
| 记忆检索不准 | 向量化质量差或阈值过低 | 人工评估检索结果 | 换embedding模型,设阈值0.7 |
| 上下文截断 | token数超模型窗口 | 打印token计数 | 滑动窗口加摘要压缩 |
| 模型忽略工具结果 | 回填格式不友好 | 检查回填文本 | 转自然语言描述 |
| 记忆冲突 | 新旧记忆未处理 | 检查更新逻辑 | 保留最新,旧版标记过期 |
提示:排查问题时先看日志。我习惯在四个模块的关键路径上都打日志,包括会话ID、消息ID、工具名称、调用耗时、记忆检索结果。出问题时按会话ID过滤日志,整条链路一目了然。
8. 一些实操后的个人体会
这套架构我前后迭代了三个版本。第一版把四个模块混在一起写,代码很快变成一团乱麻,加个新工具要改五六个文件。第二版拆分了模块但接口没设计好,模块之间耦合还是很重。第三版才把接口定清楚,每个模块只暴露必要的方法,模块间通过数据类传递信息,扩展性好了很多。
如果让我给正在做类似系统的朋友一个建议,我会说:先把会话和工具做扎实,这两个是刚需,记忆和MCP可以后加。会话做扎实的标准是并发不串号、恢复不丢消息;工具做扎实的标准是超时可控、错误可重试、结果可理解。这两个基础打好了,后面加记忆和MCP就是水到渠成的事。
另外,不要过度设计。我见过一上来就搞复杂记忆系统的,结果发现大部分用户根本不需要跨会话记忆,白白增加了系统复杂度和延迟。先跑起来,观察用户实际怎么用,再决定记忆系统要做到什么程度。这个思路帮我省了很多无用功。