1. 项目概述为什么“能跑”不等于“能用”更不等于“能扛住”“Agent进阶系列-第一课Demo能跑、上线就崩拆解底层核心壁垒”——这个标题一出来我就知道它戳中了当前大量技术实践者最真实的痛感。不是不会写代码不是不懂LLM原理而是明明在本地Jupyter里调通了API、生成了漂亮回复、连多轮对话都跑得丝滑结果一打包上测试环境延迟飙升再推到预发服务开始503等灰度放量到5%整个链路直接雪崩。这不是个别现象而是当前Agent类项目从实验室走向真实业务场景时几乎必经的“死亡之谷”。我参与过多个跨行业Agent落地项目从某高校实验室的智能导学助手到某公司内部知识问答系统再到面向终端用户的客服辅助工具无一例外在Demo阶段都显得“很聪明”但上线后暴露的问题却高度同质状态管理混乱、上下文膨胀失控、工具调用链脆弱、错误传播无隔离、可观测性为零。这些不是功能缺陷而是架构层面的结构性短板。它们藏在“能跑”的表象之下一旦流量、数据复杂度或用户交互深度突破某个阈值就会像多米诺骨牌一样接连倒下。本课不讲如何调用一个新模型API也不堆砌前沿论文术语而是聚焦于这五个最常被忽视、却决定项目生死的核心壁垒。它们不炫技但每一条都直指工程落地的命门。无论你是刚接触Agent概念的开发者还是已交付过几个Demo的技术负责人只要你的目标是让Agent真正稳定、可靠、可维护地服务于真实用户那么这堂课拆解的就是你接下来三个月最该优先投入精力的地方。2. 内容整体设计与思路拆解从“功能正确”到“系统健壮”的范式迁移2.1 为什么传统Web开发思维在Agent项目里会失效很多团队在启动Agent项目时下意识沿用过去做REST API或微服务的经验定义好输入输出写好核心逻辑加个日志上个Prometheus监控似乎就万事大吉。但Agent的本质是一个持续演化的、带状态的、多模态决策闭环。它不像一个HTTP接口一次请求一次响应边界清晰而更像一个“数字员工”需要记住对话历史、理解用户隐含意图、动态规划下一步动作、调用外部工具、处理失败重试、并在必要时主动澄清。这种根本差异导致四个关键维度的失配状态维度失配传统服务状态通常只存在于数据库或缓存中是离散、可序列化的。而Agent的状态是连续的、上下文敏感的、且高度依赖于模型推理过程本身比如token位置、attention权重分布。强行把整个对话历史塞进Redis不仅浪费内存更会导致上下文截断点错乱模型“忘记”关键前提。错误维度失配HTTP服务的5xx错误通常是明确的DB连接超时、下游服务不可达有标准重试和降级策略。而Agent的失败是模糊的模型可能生成了语法正确但逻辑荒谬的工具参数工具返回了格式合规但语义错误的结果用户一句话里混杂了三个意图模型只捕捉到一个。这类错误没有HTTP状态码可映射传统熔断器对此完全失明。可观测维度失配我们习惯看QPS、P99延迟、错误率。但对于Agent这些指标是“黑盒输出”。你看到延迟高但无法快速判断是模型推理慢、还是工具调用卡在第三方API、或是上下文太长导致token计算爆炸。缺乏对“决策链路”的追踪就像给汽车装了转速表却没装油量表和故障灯。扩展维度失配微服务通过水平扩容应对流量增长。但Agent的瓶颈往往不在CPU或内存而在上下文长度和工具调用并发度。把一个单实例Agent复制十份并不能线性提升其处理长对话的能力反而可能因共享状态库如向量数据库引发新的争用。因此本课的设计起点就是彻底放弃“把它当做一个API来运维”的旧范式转而采用“把它当做一个需要持续监护的智能体”的新范式。所有技术选型、架构设计、监控埋点都围绕这五个核心壁垒展开状态管理、上下文治理、工具编排、错误韧性、可观测性。这不是锦上添花的优化项而是从项目第一天起就必须嵌入DNA的基础能力。2.2 方案选型背后的硬核权衡为什么不用现成框架市面上已有不少Agent框架如LangChain、LlamaIndex、Semantic Kernel等它们极大降低了Demo门槛。但恰恰是这种“开箱即用”的便利掩盖了底层的复杂性。我做过一个对比实验用LangChain的ConversationalRetrievalChain搭建一个知识库问答Agent在本地跑10轮对话毫无压力但当我们将相同代码部署到K8s集群接入真实用户流量平均对话轮次15附带PDF/Excel附件解析问题立刻爆发内存占用呈指数增长GC频繁Pod反复OOM。根因在于LangChain默认将整个对话历史包括原始文档切片、嵌入向量ID、中间推理结果全部保留在Python进程内存中且缺乏对历史记录的主动裁剪和分层存储策略。所以本课选择“手撕”核心模块而非直接套用框架是基于三个不可妥协的权衡可控性优先于开发速度我们必须精确控制每一个token的生命周期。例如当用户说“参考刚才第三段提到的数据”系统必须能精准定位并加载那段历史而不是把全部20轮对话都喂给模型。这要求状态管理具备细粒度索引能力而通用框架的抽象层往往屏蔽了这种控制。确定性优先于灵活性线上服务最怕“玄学失败”。我们要求每一次工具调用其输入参数、执行环境、超时设置、重试策略都必须是可配置、可审计、可复现的。框架提供的Tool抽象虽然灵活但其内部错误处理逻辑如自动重试次数、退避算法往往是硬编码的无法满足金融、医疗等强监管场景的审计要求。可观测性内生而非外挂我们希望在Agent的每一个决策节点如“决定调用搜索工具”、“解析搜索结果”、“生成最终回复”都能打上结构化日志并关联唯一的trace_id。这需要日志埋点与核心逻辑深度耦合而不是事后在框架外加一层AOP代理。后者往往只能捕获到“调用开始”和“调用结束”丢失了最关键的“为什么调用”和“调用依据是什么”。因此本课的方案是构建一个极简但内核坚实的Agent运行时Runtime。它只有五个核心契约ContractStateStore提供带TTL、带版本、支持按语义检索的历史存储。ContextManager负责动态计算当前对话所需的最小有效上下文窗口。ToolExecutor封装工具调用的全生命周期准备、执行、校验、重试、降级。ErrorRouter根据错误类型网络、模型、工具、业务触发不同处置流重试、澄清、兜底、告警。TraceLogger为每个决策步骤生成OpenTelemetry兼容的span并注入业务上下文如用户ID、会话ID、意图标签。这五个契约构成了Agent从“能跑”迈向“能扛住”的底层骨架。下面我们就逐个拆解其实现细节与实操要点。3. 核心细节解析与实操要点穿透五个核心壁垒的实战解法3.1 壁垒一状态管理——告别内存泄漏拥抱分层持久化Agent的状态绝非简单的“对话历史列表”。它是一个多维、异构、有时效性的数据集合至少包含显式历史Explicit History用户与Agent的原始文本消息含时间戳、角色标识。隐式状态Implicit State模型在推理过程中产生的中间表示如检索到的文档片段ID、工具调用的参数快照、用户未言明但已被推断出的偏好如“用户倾向表格形式回答”。元状态Meta-State会话生命周期信息如创建时间、最后活跃时间、当前所属业务域售前/售后、用户认证等级。若将这三者全部塞进内存后果是灾难性的。我们曾在一个电商导购Agent中观察到当用户进行一场长达47轮的“比价-参数对比-优惠券查询-库存确认”对话后单个会话对象在Python内存中占用了1.2GB其中83%是重复存储的、已过期的商品SKU信息。实操解法三级状态存储架构我们采用一种受数据库Buffer Pool启发的分层存储策略存储层级数据类型存储介质TTL策略访问频率关键操作L1内存缓存最近3轮显式消息 当前决策所需隐式状态如最新检索IDPythondict/LRUCacheLRU淘汰极高每轮对话必读快速读取禁止写入大对象L2本地SSD完整显式历史 结构化隐式状态JSON Schema校验SQLite单文件WAL模式按会话ID分区TTL7天高历史回溯、澄清时读支持全文检索FTS5、按时间范围查询L3远程对象存储原始附件PDF/Excel、大尺寸嵌入向量、长期归档会话S3兼容存储如MinIO永久存档冷备低仅审计、训练数据回捞异步上传带MD5校验提示SQLite在此处并非“玩具数据库”而是经过严格压测的生产级选择。其WAL模式保证了高并发写入下的稳定性FTS5引擎提供了毫秒级的语义关键词检索能力如SELECT * FROM history WHERE messages MATCH 价格.*优惠远超纯内存方案的模糊匹配效率。我们禁用所有外键和触发器将Schema精简到极致单表结构如下CREATE TABLE sessions ( session_id TEXT PRIMARY KEY, user_id TEXT NOT NULL, domain TEXT NOT NULL, -- sales, support, hr created_at INTEGER NOT NULL, last_active INTEGER NOT NULL, status TEXT CHECK(status IN (active, archived, error)) DEFAULT active ); CREATE VIRTUAL TABLE history_fts USING fts5( session_id, role, content, timestamp, contenthistory, prefix2 3 4, -- 支持2-4字符前缀匹配加速中文分词 tokenizeunicode61 remove_diacritics 1 );关键细节与避坑心得绝不允许L1缓存存储原始附件这是最大的内存杀手。所有附件在接收时立即异步上传至L3并在L2中仅存储其唯一URI和元数据大小、类型、MD5。L1中只保留URI引用。隐式状态必须Schema化我们定义了一套严格的JSON Schema描述所有可能的隐式状态字段如retrieved_docs: array[object{id, score, source}]。每次写入L2前强制校验。这避免了因模型输出格式漂移导致的后续解析崩溃。TTL不是一刀切L2的7天TTL是针对“活跃会话”。对于标记为statusarchived的会话其L2数据会被立即迁移到L3L2中只保留一个轻量级存根stub包含会话摘要和L3 URI。这使L2数据库体积常年稳定在20GB以内。3.2 壁垒二上下文治理——让模型永远“记得重点”而非“记得全部”“上下文长度”是当前所有大模型的阿喀琉斯之踵。即使是最新的Claude 3.5 Sonnet官方宣称200K上下文但在实际长文档问答中其对文档末尾信息的召回率仍会随长度增加而显著衰减。我们的测试数据显示当输入文档超过80K token时模型对文档最后10%内容的关键事实提取准确率下降至42%。这意味着简单地把用户上传的100页PDF全文塞给模型是典型的“用力过猛效果反差”。实操解法动态上下文窗口Dynamic Context Window, DCWDCW不是一个固定大小的滑动窗口而是一个由三个策略协同驱动的智能过滤器语义相关性过滤Semantic Relevance Filter在每次模型推理前我们不将整个历史喂入而是先用一个轻量级的Embedding模型如bge-small-zh-v1.5仅120MB对L2中存储的所有历史消息进行向量化。然后计算当前用户新输入Query与每条历史消息的余弦相似度。只选取Top-KK5条最高相似度的消息作为“语义锚点”。时序重要性加权Temporal Importance Weighting语义锚点只是候选集。我们进一步为每条锚点消息计算一个“时序权重”weight 1 / (1 log2(当前时间 - 消息时间戳))。这个公式确保刚发生的对话如用户上一句的追问权重最高而三天前的初始提问即使语义相关权重也会被大幅衰减。最终我们按similarity_score * weight排序选取前N条N3构成核心上下文。结构化摘要注入Structured Summary Injection对于被过滤掉的、但业务上至关重要的历史如“用户已确认购买意向”我们不丢弃而是将其提炼为一行结构化摘要如[Intent: PurchaseConfirmed, Product: SKU-789, Time: 2024-05-20T14:22:00Z]并强制插入到上下文开头。这相当于给模型一个“记忆提示”成本极低10 token却能防止关键业务状态丢失。注意DCW的计算本身也需被监控。我们在TraceLogger中为每次DCW计算单独打一个span记录candidate_count候选消息数、filtered_count过滤后数量、avg_similarity_score平均相似度。当avg_similarity_score持续低于0.3时系统自动触发告警提示“用户对话主题发生漂移需检查意图识别模块”。关键细节与避坑心得Embedding模型必须与主模型对齐我们曾用OpenAI的text-embedding-ada-002去匹配国产模型Qwen2-72B结果发现语义相似度计算严重失真。最终我们坚持使用与主模型同源的Embedding模型如Qwen系列配套的bge确保向量空间的一致性。摘要注入不是自由发挥所有结构化摘要的模板都定义在配置中心由业务方审核。禁止出现自然语言描述如“用户好像想买”必须是机器可解析的键值对。这为后续的自动化分析如“统计本周购买意向转化率”打下基础。DCW必须可调试我们在管理后台提供一个“上下文调试视图”输入任意会话ID即可实时看到DCW为本次推理筛选出的具体消息、各自的相似度和时序权重。这是排查“模型为何忘了关键信息”的第一利器。3.3 壁垒三工具编排——从“能调用”到“调得稳、调得准、调得明”Agent的“智能”很大一部分体现在它能否恰当地调用外部工具Tool。但现实是90%的线上故障源于工具调用环节第三方API限流、工具返回格式变更、参数校验缺失、并发调用导致下游雪崩。一个未经编排的工具调用就像让一个新手司机直接上高速。实操解法声明式工具契约Declarative Tool Contract我们摒弃了框架中常见的def search(query: str) - str:这种弱类型定义转而采用YAML声明一个强约束的工具契约# tools/search.yaml name: web_search description: 在互联网上搜索与用户问题最相关的信息 input_schema: type: object properties: query: type: string minLength: 2 maxLength: 200 description: 搜索关键词必须是中文或英文 max_results: type: integer minimum: 1 maximum: 10 default: 5 required: [query] output_schema: type: array items: type: object properties: title: {type: string} url: {type: string, format: uri} snippet: {type: string} score: {type: number, minimum: 0, maximum: 1} timeout_ms: 5000 retry_policy: max_attempts: 3 backoff_base_ms: 1000 jitter_factor: 0.2 retryable_errors: [Timeout, RateLimitExceeded] fallback: type: static value: 抱歉当前搜索服务暂时不可用请稍后再试。这个YAML文件就是工具的“宪法”。它被编译为运行时可执行的验证逻辑并驱动整个调用生命周期输入校验在调用前严格校验query长度、max_results范围。任何不合规输入立即拒绝不发请求。超时控制使用asyncio.wait_for()确保单次调用绝不超5秒。超时后自动进入重试流程。智能重试重试不是简单循环。第一次失败后等待1秒第二次失败后等待1.2秒11.2第三次失败后等待1.44秒1.21.2并加入±200ms随机抖动避免重试请求同时冲击下游。优雅降级当重试耗尽不抛异常而是返回预设的fallback值。这个值可以是静态字符串也可以是另一个轻量级工具如local_knowledge_fallback。关键细节与避坑心得契约必须版本化tools/search.yaml会随业务迭代。我们要求每次修改都升级版本号如v1.2.0并存档。运行时强制校验调用方声明的版本与当前加载的版本一致。这杜绝了“开发环境用v1.1线上跑着v1.0结果参数名变了”的经典事故。fallback不是摆设我们要求每个fallback必须经过A/B测试。例如web_search的fallback文案必须在小流量上验证其用户留存率不低于主流程的80%否则视为契约不合格必须重构。工具调用必须可追溯TraceLogger为每次工具调用生成独立span其attributes中必须包含tool.name、tool.version、input.query脱敏后、output.length、is_fallback布尔值。这让我们能一眼看出是哪个工具、哪个版本、在什么输入下、是否触发了降级。3.4 壁垒四错误韧性——构建“免疫系统”而非“急救包”Agent的错误不是非黑即白的“成功/失败”而是一个光谱从轻微的“模型幻觉”生成了不存在的API端点到严重的“工具级联失败”搜索工具失败→导致总结工具无输入→最终回复为空白。传统“try-catch”只能捕获异常无法应对这种渐进式劣化。实操解法三层错误路由Three-Layer Error Router我们设计了一个基于错误语义的路由引擎将所有错误归入三大类并触发对应处置流错误大类典型场景路由策略处置流示例基础设施错误Infrastructure网络超时、数据库连接池耗尽、Redis响应慢自动重试 熔断启动Hystrix风格熔断器10秒内连续3次超时则对该服务熔断30秒熔断期间所有请求直接走fallback模型/工具错误Model Tool模型返回JSON格式错误、工具返回HTTP 400、参数校验失败澄清 重试Agent主动向用户澄清“您是想查询‘北京天气’还是‘上海天气’” 并将澄清后的明确意图重新注入DCW触发新一轮推理业务逻辑错误Business用户意图超出Agent能力范围如“帮我订机票”、政策限制如“不提供投资建议”兜底 转人工返回预设的、由法务和产品共同审核的兜底话术并在后台自动生成一个escalation_ticket包含完整会话trace推送至人工坐席队列这个路由引擎的核心是一个轻量级的规则引擎Rule Engine其规则以JSON定义{ rules: [ { id: infra_timeout, condition: error.type TimeoutError || error.code ETIMEDOUT, action: circuit_breaker, params: {service: search_api, duration_sec: 30} }, { id: model_json_parse, condition: error.type JSONDecodeError context.tool_name search, action: clarify, params: {prompt: 抱歉搜索结果格式异常。请问您具体想了解哪方面的信息} } ] }提示规则引擎的condition表达式我们选用jsonpath-ng库实现它支持复杂的嵌套JSON路径查询如$.response.headers[X-RateLimit-Remaining] 5远比正则匹配更精准、更可维护。关键细节与避坑心得错误分类必须前置我们不允许在ToolExecutor内部直接抛出Exception。所有工具执行完毕后必须返回一个标准化的ToolResult对象其中包含statussuccess/error、error_type枚举值、error_code字符串、error_message用户可见文案。ErrorRouter只消费这个对象不关心底层如何实现。这实现了错误处理与工具实现的彻底解耦。澄清不是万能的我们设置了澄清次数上限默认2次。如果用户连续两次都无法给出明确意图系统将自动触发Business类错误执行兜底转人工。避免陷入无限澄清的死循环。熔断状态必须持久化熔断器的状态open/closed/half-open不保存在内存而是写入L2的circuit_breakers表。这保证了在Pod重启后熔断状态不会丢失避免“重启即雪崩”。3.5 壁垒五可观测性——让每一次“思考”都可看见、可度量、可归因没有可观测性Agent就是一团迷雾。你看到一个错误却不知道它是发生在“检索阶段”、“工具调用阶段”还是“最终生成阶段”你看到延迟高却不知道是模型慢还是向量数据库慢还是网络慢。这使得问题排查变成一场耗时费力的侦探游戏。实操解法端到端决策追踪End-to-End Decision Tracing我们采用OpenTelemetry标准但做了关键增强将Agent的每一个逻辑决策点都建模为一个独立的Span而非仅仅包装HTTP调用。一个典型会话的Trace结构如下[Root Span: agent_inference] ├── [Span: dcw_compute] # 动态上下文计算 │ ├── [Span: embedding_query] # 语义相似度计算 │ └── [Span: temporal_weight] # 时序权重计算 ├── [Span: tool_selection] # 工具选择LLM输出解析 │ └── [Span: json_parse] # 解析LLM输出的JSON ├── [Span: tool_execute: search] # 工具执行带重试 │ ├── [Span: http_request] # 实际HTTP请求 │ └── [Span: fallback] # 若触发降级执行 ├── [Span: response_generation] # 最终回复生成 └── [Span: trace_log] # 将完整trace_id写入L2日志表供审计每个Span都注入了丰富的业务属性Attributesagent.session_id,agent.user_idagent.intent由意图识别模型预测如product_comparisonagent.step当前步骤如retrieval、tool_call、final_answerllm.model_name,llm.input_tokens,llm.output_tokenstool.name,tool.version,tool.is_fallback关键细节与避坑心得Span命名必须语义化严禁使用step_1、process_a这类无意义名称。每个Span名必须清晰表达其业务含义如vector_db_retrieve比db_query更能说明问题。Token计数必须精确我们不依赖模型API返回的usage字段它有时不准而是在发送请求前用与模型配套的Tokenizer如QwenTokenizer对输入Prompt进行精确分词计数在收到响应后对response.text进行同样分词。这让我们能精确计算“每轮对话的Token成本”是成本优化的核心依据。Trace必须可关联我们在L2的sessions表中为每一行添加trace_id字段。当运营同学在后台查到一个用户投诉“回复不准确”时只需输入session_id即可一键跳转到Jaeger UI查看该会话的完整决策链路精准定位是哪个环节出了问题。这将平均故障定位时间MTTD从小时级缩短至分钟级。4. 实操过程与核心环节实现从零搭建一个可上线的Agent Runtime4.1 环境准备与依赖安装轻量、确定、可复现我们摒弃了复杂的Docker Compose或Helm Chart采用最朴素的requirements.txtMakefile组合确保环境搭建的确定性。核心依赖如下# requirements.txt # 基础运行时 python3.11.9 uvloop0.19.0 # 替换asyncio事件循环提升I/O性能 aiofiles23.2.1 # 异步文件操作 # 核心能力 openai1.35.11 # 若使用OpenAI模型 dashscope1.17.2 # 若使用通义千问 sentence-transformers2.3.1 # Embedding模型 pysqlite3-binary0.5.2 # SQLite3增强版支持FTS5 opentelemetry-api1.24.0 opentelemetry-sdk1.24.0 opentelemetry-exporter-otlp-proto-http1.24.0 # 工具与监控 prometheus-client0.18.0 psutil5.9.8注意pysqlite3-binary是关键。标准Python自带的sqlite3模块基于系统SQLite不支持FTS5。我们必须显式安装此包并在代码中强制使用import pysqlite3 as sqlite3。这是一个容易被忽略的坑会导致本地开发正常线上部署失败。Makefile自动化# Makefile .PHONY: setup dev test deploy setup: pip install -r requirements.txt python -c import sqlite3; print(sqlite3.sqlite_version) | grep -q 3.39 || (echo ERROR: SQLite version 3.39, FTS5 not supported; exit 1) dev: # 启动本地开发服务器加载mock工具 python app.py --env dev test: # 运行单元测试覆盖DCW、ToolExecutor、ErrorRouter核心逻辑 pytest tests/ -v --covsrc/ deploy: # 构建生产镜像使用多阶段Dockerfile docker build -t my-agent-runtime:prod . docker push my-agent-runtime:prod执行make setup它会安装所有依赖检查SQLite版本是否≥3.39FTS5支持的最低版本若检查失败立即报错避免后续部署踩坑。4.2 核心模块编码StateStore与ContextManager的完整实现我们以StateStore和ContextManager为例展示核心模块的编码细节。所有代码均遵循“单一职责、高内聚、低耦合”原则。StateStore实现 (src/storage/state_store.py)import sqlite3 import json import time from typing import List, Dict, Optional, Any from contextlib import contextmanager class StateStore: def __init__(self, db_path: str): self.db_path db_path self._init_db() def _init_db(self): 初始化SQLite数据库创建表和FTS5虚拟表 with self._get_conn() as conn: # 创建会话主表 conn.execute( CREATE TABLE IF NOT EXISTS sessions ( session_id TEXT PRIMARY KEY, user_id TEXT NOT NULL, domain TEXT NOT NULL, created_at INTEGER NOT NULL, last_active INTEGER NOT NULL, status TEXT CHECK(status IN (active, archived, error)) DEFAULT active ) ) # 创建FTS5虚拟表用于消息全文检索 conn.execute( CREATE VIRTUAL TABLE IF NOT EXISTS history_fts USING fts5( session_id, role, content, timestamp, contenthistory, prefix2 3 4, tokenizeunicode61 remove_diacritics 1 ) ) # 创建普通history表存储结构化数据 conn.execute( CREATE TABLE IF NOT EXISTS history ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT NOT NULL, role TEXT NOT NULL CHECK(role IN (user, assistant)), content TEXT NOT NULL, timestamp INTEGER NOT NULL, metadata TEXT, -- JSON string for implicit state FOREIGN KEY(session_id) REFERENCES sessions(session_id) ) ) conn.commit() contextmanager def _get_conn(self): 获取线程安全的数据库连接 conn sqlite3.connect(self.db_path, check_same_threadFalse) conn.row_factory sqlite3.Row try: yield conn finally: conn.close() def save_message(self, session_id: str, role: str, content: str, metadata: Optional[Dict[str, Any]] None): 保存一条消息到L2同时更新会话状态 now int(time.time()) with self._get_conn() as conn: # 插入到history表 conn.execute( INSERT INTO history (session_id, role, content, timestamp, metadata) VALUES (?, ?, ?, ?, ?), (session_id, role, content, now, json.dumps(metadata) if metadata else None) ) # 插入到FTS5虚拟表用于检索 conn.execute( INSERT INTO history_fts (session_id, role, content, timestamp) VALUES (?, ?, ?, ?), (session_id, role, content, now) ) # 更新sessions表的last_active conn.execute( INSERT OR REPLACE INTO sessions (session_id, user_id, domain, created_at, last_active, status) VALUES (?, (SELECT user_id FROM sessions WHERE session_id?), (SELECT domain FROM sessions WHERE session_id?), (SELECT created_at FROM sessions WHERE session_id?), ?, active), (session_id, session_id, session_id, session_id, now) ) conn.commit() def get_recent_messages(self, session_id: str, limit: int 10) - List[Dict]: 获取最近N条消息用于L1缓存填充 with self._get_conn() as conn: rows conn.execute( SELECT role, content, timestamp, metadata FROM history WHERE session_id ? ORDER BY timestamp DESC LIMIT ?, (session_id, limit) ).fetchall() return [dict(row) for row in rows] def search_messages(self, session_id: str, query: str, limit: int 5) - List[Dict]: 在指定会话内进行语义关键词搜索 with self._get_conn() as conn: # 使用FTS5的MATCH语法 rows conn.execute( SELECT session_id, role, content, timestamp FROM history_fts WHERE session_id ? AND history_fts MATCH ? ORDER BY rank LIMIT ?, (session_id, query, limit) ).fetchall() return [dict(row) for row in rows]ContextManager实现 (src/context/context_manager.py)import time import math from typing import List, Dict, Tuple, Optional from sentence_transformers import SentenceTransformer from src.storage.state_store import StateStore class ContextManager: def __init__(self, state_store: StateStore, embedding_model: str bge-small-zh-v1.5): self.state_store state_store self.embedding_model SentenceTransformer(embedding_model) # 缓存Embedding向量避免重复计算 self._embedding_cache {} def _get_or_compute_embedding(self, text: str) - List[float]: 获取或计算文本的Embedding向量