AI工程SynapseB类

Synapse 案例设计:把 Claude Code 接入有状态管线的研发场景

用任务拓扑层 + 状态快照解决 Claude Code 在多步管线中的上下文断裂 —— 不是「完全自主」,是把人类不该做的环节交回机器

  • Synapse用任务拓扑编排替代Claude Code的单Agent对话循环
  • 多轮迭代中的上下文丢失是自研Agent框架的核心陷阱
  • 用「状态快照 + 阶段门控」可稳定80%以上的幻觉问题
  • Claude Code直接跑复杂场景需手动维护session状态
  • Synapse的checkpoint机制让任务可回滚而不重跑

问题背景

我们团队在去年Q4启动了一个内部项目:基于Claude Code构建一个自动化代码审查Agent。目标听起来很简单——让Claude Code自动扫描仓库、生成审查报告、并通过飞书机器人推送给对应负责人。

第一版用了3周跑通,P0级Bug为0。但当业务方提出新需求「按分支维度聚合报告」时,问题开始出现:Claude Code在连续多轮对话中开始出现上下文跳跃——有时候生成的diff摘要漏掉文件,有时候报告格式在第4轮突然漂移。一周内堆了17个issue,其中9个指向「Agent状态不一致」。

这不是Claude Code本身的问题。是我们把Claude Code当成了主力推理引擎,却没有为它设计状态管理层。就像让一个记忆力很强的助手在没有笔记本的情况下做连续会议记录——他不会忘记,但他没有地方「记下来」供下一个任务查阅。

为什么这个决策难做

我们一开始以为问题出在prompt上——"是不是指令不够清晰?""是不是没有加few-shot示例?"所以花了2天反复调整system prompt,加入了3个参考示例,但测试结果显示:上下文跳跃依然在第3轮后开始出现,只是概率从78%降到了61%。

但实际上真正的问题在于:Claude Code的对话窗口是Session级的,而我们的代码审查流程是Pipeline级的。一个审查任务包含「拉取diff → 过滤测试文件 → 生成摘要 → 格式化报告 → 推送通知」5个步骤,每个步骤的输入都依赖上一个步骤的输出。我们在用Claude Code处理一个需要「状态传递」的流水线的场景,却把它当作一个问答对话来用。

这导致每个步骤的Claude Code调用都是"新鲜"的——它不知道上一个步骤过滤掉了哪些文件,因为那个信息只存在于我们没有显式存储的地方。这才是上下文跳跃的根因:不是模型"忘记"了,是设计者没有给它建立记忆机制。

Claude Code的差异化优势在于代码生成和调试能力,但用它直接驱动多步骤Pipeline时,必须显式管理状态流——这是它不会替你做的设计决策。

根因:缺少任务拓扑层导致的上下文断裂

问题的本质是我们把Claude Code当成了「全知助手」,而实际上它是一个「专注执行器」。在Synapse的架构设计中,我们将这个关系倒转了:Synapse负责任务拓扑(Topology),Claude Code负责每个节点的具体执行。

具体来说,我们在Synapse中设计了一个task_state快照机制。每个Pipeline节点完成后,会将关键状态写入一个共享的JSON context:

# Synapse Pipeline State Manager(简化版核心逻辑)
import json
import hashlib

class TaskState:
    def __init__(self, task_id: str):
        self.task_id = task_id
        self.context = {
            "metadata": {"version": "1.0", "task_id": task_id},
            "steps": {}
        }
        self.checkpoints = []

    def record_step(self, step_name: str, output: dict, artifacts: list):
        """每个Pipeline步骤完成后调用,固化输出状态"""
        step_record = {
            "step": step_name,
            "output_hash": hashlib.sha256(
                json.dumps(output, sort_keys=True).encode()
            ).hexdigest()[:8],
            "artifacts": artifacts,
            "input_for_next": self._derive_next_input(step_name, output)
        }
        self.context["steps"][step_name] = step_record
        self.checkpoints.append({
            "timestamp": self._now(),
            "step": step_name,
            "snapshot": output.copy()
        })

    def _derive_next_input(self, step_name: str, output: dict) -> dict:
        """从当前输出推导下一节点的输入契约"""
        next_inputs = {
            "filter_files": {"files_to_review": output["filtered_files"]},
            "generate_summary": {"diffs": output["diffs"]},
            "format_report": {"summary": output["summary"]},
            "notify": {"report": output["report"]}
        }
        return next_inputs.get(step_name, {})

    def get_checkpoint(self, step_name: str):
        for cp in self.checkpoints:
            if cp["step"] == step_name:
                return cp["snapshot"]
        return None

关键设计逻辑在_derive_next_input函数:每个步骤完成后,我们会从它的输出中提取「下一节点的输入契约」,而不是让Claude Code自己去推断。相当于在Pipeline层面建立了一个「交接清单」——A步骤交接什么、B步骤接收什么,都是显式定义的。

当我们把这个机制引入Synapse的工作流引擎后,Claude Code的每次调用都带着明确的「本步骤输入」和「期望输出格式」。上下文跳跃的概率从61%降到了4%以下。

可移植的原则

  1. 如果你在构建多步骤AI Pipeline,不要依赖模型自身的对话记忆来传递状态——显式设计状态快照层,用结构化数据(JSON/msgpack)替代隐式上下文。
  2. 如果你在用Claude Code处理超过3轮的连续任务,在每个节点之间插入状态固化checkpoint——记录当前输出hash和关键字段,用于下游节点显式读取,而非让模型重新"理解"。
  3. 如果你发现Claude Code输出格式不稳定,先检查是否在用「问答模式」处理「Pipeline任务」——这往往是根本性的架构错配,而非prompt调优问题。
  4. 如果你需要任务可回滚,让checkpoint快照包含完整的输入输出对——Synapse的checkpoint机制让你可以从任意步骤重新注入状态,而不需要重新运行整个流程。

结尾

回到文章开头提到的那个代码审查Agent案例。用Synapse重构后的Pipeline,状态管理逻辑从200行手工维护的session字典,变成了不到80行的结构化State类。新增「按分支聚合」需求时,我们只需要在_derive_next_input里增加一条映射关系,Claude Code端不需要任何改动。这才是Synapse真正带来的差异:不是让Claude Code更强,而是让它在一个有结构的执行框架里发挥确定性的能力。

如果你正在构建类似的复杂多步骤Agent系统,并且还在用"反复调prompt"的方式解决上下文问题——建议先停下来检查你的Pipeline是否有状态管理层。没有状态管理的Claude Code,就像一台没有内存条的高性能电脑:跑得动,但跑不久。