Finn Days
博客笔记项目标签关于
博客笔记项目标签关于
Finn Days
博客·关于·RSS

用中间件模式约束 Agent:一个生产级 Harness 框架的设计

2026.02.08·9 min read
AgentHarness EngineeringMiddlewareLangGraph

问题

LLM Agent 在 demo 中表现得很好,部署到生产环境后会暴露一系列可靠性问题。以下是实际遇到的几种典型异常:

Agent 在生产中的四种典型异常

  1. 不检索就编答案 — 用户问产品参数,Agent 直接输出一个看似合理但完全编造的数字
  2. 无限循环搜索 — 搜"张工 工作"→"张工 工作内容"→"张工 工作进展",本质是同一个 query,但 Agent 停不下来
  3. 多轮对话丢上下文 — 第一轮讨论运营商网络部署,第二轮用户说"只看联通的",Agent 不知道在筛选什么
  4. 超时后返回残缺回复 — 做了大量检索但来不及整合,给用户返回空字符串或"我需要更多信息"

这些问题的共同点是:不能靠调 prompt 解决。你在 system prompt 里写"必须先搜索",Agent 大概率会遵守——但不是 100%。而生产环境需要的是 100%。

框架总览

我们的方案是在 LLM 调用前后插入一系列中间件,用确定性的代码逻辑约束概率性的模型行为。整个框架的结构如下:

      • builder.py
      • force_first_search.py
      • min_tool_calls.py
      • max_tool_calls.py
      • stale_message_cleanup.py
    • service.py
    • builder.py
    • session_manager.py
    • config.py

每次 generate_reply() 调用都经过以下流程:

用户 Query
  → SessionManager.resolve_session()     冷/热判断 + 话题检测 + Query 改写
  → MemoryRetriever.retrieve()           长期记忆检索(3s 超时降级)
  → build_deep_agent()                   组装 system prompt + 中间件链 + 工具
  → agent.invoke()                       LangGraph 执行(含中间件拦截)
  → MessageExtractor                     提取当前轮 AI 回复
  → ResponseFormatter                    清理中间件残留文本
  → AnswerConsolidator                   答案质量检测 + 不完整补救

中间件链:设计与执行顺序

middleware/builder.py
middleware = [
    # 0. 必须第一个:清理上轮残留的中间件指令
    StaleMessageCleanupMiddleware(),
 
    # 1. 强制首次搜索
    ForceFirstSearchMiddleware(required_tool="knowledge_search", max_injection_count=2),
 
    # 2. 最小工具调用次数
    MinToolCallsMiddleware(min_search_calls=2, min_total_calls=3),
 
    # 2.5 最大工具调用(安全网)
    MaxToolCallsMiddleware(max_tool_calls=25, max_search_calls=15),
 
    # 3. 上下文编辑(清理早期工具输出,防 token 溢出)
    ContextEditingMiddleware(trigger=48000, keep=10),
 
    # 4. 上下文摘要压缩
    SummarizationMiddleware(model=model, trigger=54400, keep=12800),
 
    # 5. 模型降级
    ModelFallbackMiddleware(fallback_model_1, fallback_model_2),
]

StaleMessageCleanup 为什么必须排第一?

多轮对话中,checkpointer 恢复上一轮的完整 state,其中包含 ForceFirstSearch 上一轮注入的"必须先搜索"SystemMessage。如果不先清理,新一轮的 ForceFirstSearch 看到这条残留指令会误判"已经注入过了",跳过强制搜索——恰恰和设计意图相反。

ForceFirstSearch 的消息序列

这个中间件通过 4 个 hook 工作。以下是一次完整的消息序列对比:

══════ 无 ForceFirstSearch ══════

  HumanMessage: "R6900 G5 支持多大内存?"
  AIMessage: "R6900 G5 最大支持 512GB 内存"     ← Agent 直接编造了答案

══════ 有 ForceFirstSearch ══════

  HumanMessage: "R6900 G5 支持多大内存?"

  [before_model 第1次] 检测到首次模型调用且未搜索
    → 注入 SystemMessage: "【强制检索要求】回答前必须先调用 knowledge_search..."

  AIMessage: tool_calls=[{name: "knowledge_search", args: {question: "R6900 G5 内存规格"}}]

  [after_model] 检测到 tool_calls 包含 knowledge_search
    → 标记 _has_searched = True,后续不再干预

  ToolMessage: {status: "success", results: [{title: "R6900 G5 产品规格", content: "最大支持 1TB DDR5..."}]}

  AIMessage: "根据产品文档,R6900 G5 最大支持 1TB DDR5 内存 [参考来源:R6900 G5 产品规格]"

如果 Agent 在第一次注入后仍然不调用工具(直接回答),中间件会再注入一次更强硬的提醒。最多 max_injection_count=2 次后放弃,避免无限注入撑爆 context。

上下文压缩的分级策略

触发条件: token 数超过 max_input_tokens × 0.75(约 48000 tokens)

策略: 直接删除早期的 ToolMessage(工具输出通常是最大的 token 消耗者,一次搜索返回 5 个文档约 4000 tokens),只保留最近 10 个。被删除的内容替换为占位符。

优势: 零延迟(纯规则,不调 LLM),释放空间最明显。

两者是分级协作:ContextEditing 先清理工具输出(成本最低、效果最明显),如果仍然超限再触发 Summarization 做更精细的压缩。

会话管理:冷启动 vs 热继续

触发条件: checkpointer 中无 session,或话题检测判定为切换。

  • 新建 session(bump version → 新 thread_id)
  • 注入虚拟文件(skills、references)到 Agent state
  • 将历史摘要注入 system prompt
  • 对 Query 做简单规则改写(个人代词替换:"帮我查" → "帮张工查")

话题检测用两级策略:第一级,关键词快速路径——"重新查"、"不对"、"错了"、"接着说"这些一定不是话题切换,零延迟返回。第二级,用轻量 LLM(temperature=0)判断当前 query 与最近 5 条消息是否同一话题。

上下文感知改写是热继续的关键。用户说"只分析联通"时缺少上下文,改写逻辑检测到这是过滤筛选类表达后,自动补全为"帮我分析三大运营商的5G核心网部署情况,只分析联通"。

异常恢复:从半成品中抢救答案

Agent 超时或递归超限(GraphRecursionError)时,不能直接返回错误——用户已经等了几十秒。

策略一(优先):从工具结果整合

从 checkpointer 的 partial state 中提取所有成功的 ToolMessage,用一个独立的 LLM 调用重新生成回答。

通常 Agent 超时是因为在整合阶段卡住,但之前的检索结果是好的——用这些结果重新生成比什么都没有强得多。实测成功率 ~70%。

策略二(回退):提取最后的 AI 回复

如果策略一失败(没有可用的工具结果),检查 Agent 在超时前是否已经开始输出回复。如果最后的 AIMessage 内容 ≥100 字,直接截取作为回复——虽然不完整,但比空好。

答案质量兜底

Agent 有时返回技术上"成功"但内容残缺的回答。AnswerConsolidator 做最后一道检查:

answer_consolidation.py
# 检测模式
INCOMPLETE_PATTERNS = ["已为您整理", "已为您梳理", "如需了解"]
MIDDLEWARE_LEAKAGE = ["【强制检索要求】", "【系统指令】"]
 
def is_incomplete_answer(answer, tool_calls_count):
    # 有工具调用但回答太短 → 内容可能丢失
    if tool_calls_count >= 4 and len(answer) < 150:
        return True, "搜了4次但回答不到150字"
    # 包含中间件残留文本
    if any(p in answer for p in MIDDLEWARE_LEAKAGE):
        return True, "回复混入了中间件指令"
    return False, ""

检测到不完整时,从消息历史中提取 ToolMessage 重新整合——和超时恢复的策略一相同。这是保守策略,只在明显异常时触发。

Harness Engineering 视角

2026 年 OpenAI 发表了 Harness Engineering 论文后回头看,我们的框架和它的理念高度一致。Harness 是围绕 LLM Agent 的一切控制系统——不是模型本身,而是让模型可靠工作的"缰绳"。

对应到我们的系统:

Harness 层我们的实现
Context Engineering7 层 system prompt 组装 + SummarizationMiddleware
GuardrailsForceFirstSearch + MinToolCalls + MaxToolCalls
State ManagementSessionManager + RedisSaver Checkpointer
Recovery_recover_from_partial_state + AnswerConsolidator
ObservabilityLangfuse ProcessTracer(关键节点 Span 追踪)

核心收获

中间件是整个框架中价值最高的设计。它把"Agent 应该怎么做"从概率性的 prompt 指令变成了确定性的代码约束——不是在 prompt 里写"你必须先搜索"(Agent 可能忽略),而是在代码层检测到没搜索就注入系统消息强制要求。这比单纯调 prompt 可靠一个数量级。

Harness Engineering: Leveraging Codex in an Agent-First World

OpenAI 关于 Harness Engineering 的论文,核心观点:竞争优势不在模型,而在控制系统。

openai.com

Deep Agents Overview — LangChain

LangChain 的 Deep Agents 框架文档,包含中间件、虚拟文件系统和子代理的设计。

docs.langchain.com

上一篇

企业知识库的混合检索架构:三路融合与精排实践

下一篇

生产环境中 Agent 记忆系统的设计与缺陷

分享:··

Finn

AI 工程师,做 Agent 系统和 RAG 检索。在这里写下工程实践中踩过的坑和学到的东西。

订阅 Newsletter

每周精选技术文章,直达你的收件箱。不发垃圾邮件,随时退订。

我们尊重你的隐私,邮箱信息仅用于发送 Newsletter,不会分享给第三方。

评论

目录

  • 问题
  • 框架总览
  • 中间件链:设计与执行顺序
    • ForceFirstSearch 的消息序列
    • 上下文压缩的分级策略
  • 会话管理:冷启动 vs 热继续
  • 异常恢复:从半成品中抢救答案
  • 答案质量兜底
  • Harness Engineering 视角