1. 项目概述：当“智能体开发”撞上真实钱包厚度

你有没有试过在深夜调试一个本该自动跑完的代码生成流程，结果发现它卡在了第三步——不是因为逻辑错了，而是因为调用的某个外部API突然返回了429（Too Many Requests），而你的 orchestrator 根本没意识到这个错误会像多米诺骨牌一样，让后续所有依赖它的子任务全部失效？更糟的是，你刚花30分钟手动重跑失败链路，结果发现上游某个被人类审核过的中间结果其实有歧义，导致下游模型反复生成错误结构……这种“看似智能、实则脆弱”的体验，正是当前绝大多数 Agentic 系统的真实写照。而这篇内容要讲的，就是我如何用一台二手 ThinkPad T14（i5-1135G7 + 16GB RAM）、一个 €40 的年度云服务预算（不含硬件），从零搭建出一套 真正扛得住现实干扰 的智能体协作系统——它不追求炫技的多智能体辩论，也不堆砌昂贵的向量数据库，而是把“依赖感知”和“人机协同节奏”刻进每一行调度逻辑里。核心关键词是： Reliable Agentic Development、Dependency-Aware Orchestration、Claude、Codex、Human-in-the-Loop 。它适合三类人：一是正在用 LangChain/LlamaIndex 做 PoC 却总被线上稳定性劝退的工程师；二是想把 AI 工具链嵌入现有业务流程（比如法务合同初筛、电商客服知识库更新）但又不敢全自动化的产品经理；三是手头只有学生优惠额度或小团队云预算，却需要交付可长期运行系统的独立开发者。它解决的不是“能不能做”，而是“做了之后敢不敢关掉监控告警”。

这个项目不是在教你怎么调用 Claude 的 API，也不是教你如何微调 Codex 模型——那些文档里都有。它解决的是文档里 绝不会写 的问题：当 Claude 在处理一份 PDF 合同时，突然因 token 超限被截断，而 Codex 正等着它的结构化 JSON 输出去生成测试用例，此时系统该立刻重试、降级为摘要模式、还是直接触发人工介入？谁来判断？依据什么？这个决策过程本身，就是“依赖感知编排”的心脏。€40 不是上限，而是倒逼你剔除所有华而不实的抽象层，直面真实世界里的资源约束、网络抖动、模型不确定性与人类认知带宽限制。我试过把整个流程跑通 17 次，前 12 次都倒在同一个地方：当人类审核员在 Slack 里回复 “这个条款需要法务复核” 时，系统要么等他等到超时（浪费钱），要么强行跳过（埋下风险）。直到我把“人类响应预期时间”作为一个显式参数写进任务图谱，问题才真正消失。这背后没有魔法，只有一套清晰的状态机、一个轻量级的持久化队列，和对每个环节失败成本的冷酷计算。

2. 整体架构设计：为什么放弃 LangGraph，选择自研状态机驱动的 DAG 调度器

2.1 核心思路：把“可靠性”拆解为可测量、可干预的四个维度

很多团队一上来就选 LangGraph 或 LlamaIndex 的 Agent 框架，觉得“官方出品，肯定可靠”。我试过，也踩过坑。LangGraph 的 StateGraph 确实优雅，但它默认把“状态”当作一个扁平字典，所有节点共享同一份 mutable state。这意味着：当 Claude 节点正在解析合同时，Codex 节点如果误读了尚未写入完成的中间字段，就会拿到脏数据。更致命的是，它的错误恢复机制是“重放整个图”，而不是精准回滚到故障节点上游。一次网络抖动导致的 429 错误，可能让你白白烧掉 €0.8 的 API 费用去重跑所有前置步骤——而这 €0.8，在 €40 预算里占了 2%。所以我的第一原则是： 可靠性 = 可预测的失败成本 + 可控的恢复粒度 + 显式的依赖契约 + 人类干预的零摩擦入口 。这四个维度，必须在架构层面硬编码，不能靠文档约定。

因此，我彻底放弃了通用 Agent 框架，转而构建一个极简的、基于内存状态机 + SQLite 持久化的 DAG 调度器。它只有三个核心实体： TaskNode （定义输入/输出 schema、执行函数、重试策略）、 DependencyEdge （声明 A 节点的 output.field_x 必须作为 B 节点的 input.field_y）、 HumanGate （一个特殊节点，其执行函数是“发消息给指定 Slack channel 并等待回复”）。整个图谱不是在运行时动态生成的，而是在部署前用 YAML 静态定义的。例如，一份合同分析流程的图谱文件 contract_review.yaml 会明确写出：

nodes:
  - id: "parse_pdf"
    type: "claude"
    input_schema: {pdf_url: "string"}
    output_schema: {raw_text: "string", page_count: "integer"}
    retry_policy: {max_attempts: 3, backoff: "exponential"}

  - id: "extract_clauses"
    type: "codex"
    input_schema: {text: "string"}
    output_schema: {clauses: "[{type: string, content: string}]"}
    # 关键：这里声明它依赖 parse_pdf 的 raw_text 字段
    dependencies: ["parse_pdf.raw_text"]

  - id: "human_review"
    type: "human_gate"
    input_schema: {clauses: "[{type, content}]"}
    # 它不依赖 extract_clauses 的全部输出，只依赖其中 type=="payment" 的子集
    dependencies: ["extract_clauses.clauses[?(@.type=='payment')]"]

看到没？ dependencies 字段不是简单的 "parse_pdf" ，而是精确到字段级的 JSONPath 表达式。这意味着调度器在 extract_clauses 执行前，会先检查 parse_pdf 的输出中是否存在 raw_text 字段，且类型为 string。如果 parse_pdf 因 token 超限只返回了 {page_count: 12} ，调度器会立刻标记该节点为 PARTIAL_FAILURE ，而不是让下游盲目执行。这个设计直接解决了“依赖感知”的第一层： 数据契约的显式校验 。

2.2 为什么 SQLite 是 €40 预算下的最优解？一个被低估的持久化选择

你可能会问：为什么不用 Redis？它更快啊。或者用 PostgreSQL？它更健壮啊。答案很实在： Redis 的内存成本在 €40 预算里是奢侈品，PostgreSQL 的管理开销会吃掉你 30% 的运维时间 。我做过测算：一个中等复杂度的合同分析流程，单次执行会产生约 8KB 的中间状态（含日志、schema 校验结果、重试计数）。按每天 200 次调用算，一年就是 576MB。SQLite 的单文件数据库，存这个量级的数据，读写延迟稳定在 3ms 内，且完全无需单独进程、无需连接池、无需备份脚本——你只要把那个 .db 文件放在云服务器的 /var/data/agent.db ，它就永远在线。更重要的是，SQLite 支持 WAL（Write-Ahead Logging）模式，允许多个线程安全地并发读写，这完美匹配了我们“一个调度器进程 + 多个异步 worker”的架构。

我对比过三种方案的实际开销：

方案	年度成本（€）	首次部署时间	故障恢复时间	数据一致性保障
Redis（Hobby Tier on Render）	72	<5 分钟	<10 秒（但需额外写哨兵脚本）	弱（无事务，AOF 可能丢最后几条）
PostgreSQL（Supabase Free Tier）	0（但有连接数限制）	25 分钟（配 SSL、role、policy）	3-5 分钟（需手动查 pg_stat_activity）	强（ACID）
SQLite（本地文件 + WAL）	0	<2 分钟	0 秒（进程重启即恢复）	强（ACID，且 WAL 保证崩溃安全）

注意看最后一行：SQLite 的“故障恢复时间”是 0 秒。因为它的状态就是文件本身，没有“连接断开”概念。当调度器进程因 OOM 被 kill，你只要 systemctl restart agent-scheduler ，它会从 SQLite 里读取上次保存的 task_status 和 last_updated 时间戳，自动续跑。这比任何分布式协调服务都更符合“€40 可靠性”的本质—— 用最朴素的工具，实现最确定的行为 。当然，它有局限：不支持水平扩展。但这恰恰是好事。€40 预算下，你本就不该设计成需要水平扩展的系统。如果流量真大到 SQLite 瓶颈，说明你已经成功了，该去申请更大预算了。

2.3 Claude 与 Codex 的角色切割：不是“谁更强”，而是“谁更稳”

很多人纠结该用 Claude 还是 Codex 来做代码生成。我的结论很反直觉： 在可靠性优先的场景下，Codex（具体指 code-davinci-002 ）比 Claude 3 Haiku 更值得信赖 。不是因为 Codex 更聪明，而是因为它更“笨”、更可预测。Codex 是一个纯粹的代码补全模型，它的输入输出格式极其固定：你给它一段 Python 注释 + 函数签名，它就还你一段 Python 实现。而 Claude 3 Haiku 虽然快，但它是一个通用对话模型，当你让它“生成一个测试用例”，它可能返回 Markdown 表格、JSON、甚至一段解释性文字——这直接破坏了我们前面定义的 output_schema 契约。

所以我的角色分配是铁律：

• Claude 负责“理解”与“结构化” ：PDF 文本解析、合同条款分类、模糊需求澄清。它输出必须是严格 JSON，且 schema 由 pydantic.BaseModel 强校验。
• Codex 负责“执行”与“生成” ：根据 Claude 输出的结构化条款，生成对应单元测试、SQL 查询、API 文档片段。它的 prompt 里永远包含 Output only valid JSON. No explanations. 这句话，并在调度器里加一层正则校验： if not re.match(r'^\s*\{.*\}\s*$', output): raise SchemaViolation("Non-JSON output") 。

这个切割带来了两个实际好处：第一，当 Codex 偶尔“发疯”输出乱码时，校验层会在 50ms 内捕获并触发重试，不会污染下游；第二，Claude 的 token 消耗可以被精准预估。比如，一份 10 页 PDF，Claude 解析后平均输出 1200 tokens 的 JSON，那么我就可以在调度器里设置 max_tokens: 1500 ，一旦实际消耗超过此值，立即终止并标记为 TOKEN_OVERFLOW ，而不是让它硬着头皮截断——这避免了下游拿到半截 JSON 导致解析崩溃。Codex 则相反，我给它 max_tokens: 500 ，因为它的输出长度非常稳定，几乎从不触发截断。这种“用模型的确定性，对抗世界的不确定性”，才是 €40 预算下真正的工程智慧。

3. 核心细节解析：Human-in-the-Loop 不是加个按钮，而是设计一个“人类响应 SLA”

3.1 HumanGate 节点的三重状态机：从“发消息”到“收确认”的完整闭环

把人类塞进自动化流程，最大的陷阱是把它当成一个“黑盒阻塞点”。很多方案只是简单地在流程里插一个 input() 函数，然后让程序挂起。这在本地测试没问题，但在生产环境里，等于主动放弃可靠性——如果审核员忘了回复，整个流水线就永久卡死。我的 HumanGate 节点，是一个拥有完整生命周期的状态机，它有且仅有三种状态：

1. PENDING ：已向 Slack 发送消息，等待回复。此时记录 sent_at: timestamp 。
2. ACKNOWLEDGED ：收到审核员在 Slack 中的任意回复（哪怕只是“ok”），记录 ack_at: timestamp ，并进入人工处理阶段。
3. RESOLVED ：审核员通过预设的 /resolve Slash Command 提交结构化结果（如 {"status": "approved", "comments": "payment term ok"} ），调度器验证 JSON 合法性后，将结果写入 SQLite，并触发下游。

关键在于， PENDING 状态不是无限期的。我在节点定义里强制要求 sla_seconds: 1800 （30 分钟）。调度器有一个独立的 SLAWatcher 线程，每 60 秒扫描一次 SQLite，找出所有 status == 'PENDING' and sent_at < now() - sla_seconds 的任务，并自动执行降级策略。这个降级策略不是“跳过”，而是 根据业务语义决定下一步 。例如，在合同审查中， sla_seconds: 1800 的降级动作是：“自动将该条款标记为 NEEDS_LEGAL ，并通知法务组负责人，同时允许下游生成‘待法务确认’版本的测试用例”。你看，它没有破坏流程，而是把“人类不可靠”这个事实，转化为了一个可追踪、可审计、可补偿的业务状态。

提示：Slack 的 Slash Command /resolve 不是魔法。它背后是一个极简的 Flask Webhook，接收请求后，只做三件事：1) 验证 Slack 签名（防伪造）；2) 解析 JSON payload；3) 向 SQLite 的 human_responses 表插入一行。整个过程不到 120ms，且不依赖任何外部服务。我把这个 Webhook 部署在同一台 VPS 上，用 Nginx 反向代理，连 HTTPS 都省了（内网通信，够用）。

3.2 依赖感知的“动态重试”：不是次数越多越好，而是越准越好

传统重试策略（如指数退避）有个致命缺陷：它假设每次失败的原因相同。但现实中，Claude 的 429 错误（限流）和 Codex 的 500 错误（服务器宕机），需要的应对方式完全不同。前者应该“稍等再试”，后者应该“立刻换模型”。我的调度器实现了“错误码感知重试”（Error-Code-Aware Retry）。它在每次 API 调用后，不仅捕获异常，还深度解析响应体：

• 对于 Claude：检查 response.headers.get('x-ratelimit-remaining') 和 response.json().get('error', {}).get('type') 。如果是 rate_limit_exceeded ，则应用 backoff: exponential, base_delay: 2s, max_delay: 30s ；如果是 invalid_request_error （如 token 超限），则立即降级：截断输入文本，添加 ... [TRUNCATED] 标记，并重试。
• 对于 Codex：检查 response.status_code 和 response.text.startswith('{"error":') 。如果是 500 ，则切换到备用 endpoint（我注册了两个 Codex key，主 key 用 code-davinci-002 ，备 key 用 code-cushman-001 ，后者更慢但更稳）；如果是 400 ，则检查是否因 max_tokens 设置过小，自动增加 100 tokens 并重试。

这个逻辑写在 retry_strategy.py 里，只有 87 行代码，但它让重试成功率从 63% 提升到 92%。更重要的是，它让每一次重试都有明确的“业务意图”，而不是盲目地“再试一次”。比如，当 parse_pdf 节点因 token 超限失败，调度器不会傻等 2 秒后重试——它会立刻启动一个 truncate_and_summarize 子流程：用 Claude 先生成一页摘要，再把摘要喂给 Codex 去提取关键条款。这个子流程的输出 schema 是 {summary: string, key_clauses: [...]} ，它和原流程的输出不同，但足以支撑下游的“快速通道”模式。这就是“依赖感知”的高阶形态： 当主依赖失效，能基于对下游需求的理解，提供一个语义等价的替代依赖 。

3.3 €40 预算的硬约束如何倒逼出极致的 Token 管理

Claude 和 Codex 的账单，是 €40 预算里最不可控的部分。我最初的 PoC，一天就烧掉了 €3.2，全是 token 浪费。问题出在三个地方：1) Claude 解析 PDF 时，把整篇文档不分青红皂白喂进去；2) Codex 生成测试用例时，prompt 里塞了 500 行无关的上下文；3) HumanGate 的 Slack 消息，把整个 JSON 输出原样贴过去，审核员根本懒得看。

解决方案是“三层 Token 截断”：

• 第一层：输入预过滤（Input Pre-Filtering） 。在 parse_pdf 节点执行前，先用 pypdf 提取 PDF 文本，然后用 nltk 的 sent_tokenize 拆句，再用一个极简的 TF-IDF 向量（只保留 top 500 个词）计算每页与“contract”, “payment”, “liability” 等关键词的相似度。只保留相似度 > 0.3 的页面。实测下来，10 页合同平均只传 3.2 页给 Claude，token 消耗降 68%。
• 第二层：Prompt 压缩（Prompt Compression） 。Codex 的 prompt 不是静态字符串，而是一个 Jinja2 模板。模板里有 {% if clause.type == 'payment' %}...{% endif %} 这样的条件块。调度器在渲染前，会先分析 clause 对象的字段，只展开真正需要的分支。一个包含 5 类条款的 JSON，最终生成的 prompt 平均只有 210 tokens，而不是原先的 890。
• 第三层：人类界面精简（Human Interface Simplification） 。Slack 消息绝不发原始 JSON。而是用 tabulate 库生成一个 ASCII 表格：

Payment Terms (Page 7):
┌──────────────┬────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────......

审核员一眼就能抓住重点，回复 /resolve {"status": "approved"} 的平均时间从 4.2 分钟降到 58 秒。这不仅省了钱，更提升了整个流程的可靠性——人类响应越快，SLA 越容易满足。

4. 实操过程：从零部署一个可运行的合同审查流水线

4.1 环境准备与依赖安装：为什么只用 Python 3.11 和 7 个包

我的整个系统，只依赖 7 个 PyPI 包，且全部是纯 Python 或带预编译 wheel 的：

pip install \
  pypdf==3.17.2 \
  nltk==3.8.1 \
  tabulate==0.9.0 \
  pydantic==2.6.4 \
  requests==2.31.0 \
  flask==2.3.3 \
  apscheduler==3.10.4

注意：我 刻意避开了 langchain , llama-index , openai （官方 SDK）这些“大而全”的包。原因有三：第一，它们引入了大量间接依赖（如 tenacity , httpx , pydantic<2.0 ），增加了版本冲突风险；第二，它们的抽象层会掩盖底层 API 的真实行为（比如 openai SDK 会自动重试 429，但不告诉你它重试了几次，这违反了我们对“失败成本可预测”的要求）；第三，它们的文档假设你用的是 OpenAI 官方 endpoint，而我需要同时对接 Anthropic（Claude）和 Azure OpenAI（Codex），必须自己管理 auth header 和 endpoint routing。

所以，所有 API 调用都用原生 requests 。Claude 的调用函数长这样：

def call_claude(messages: List[Dict], model: str = "claude-3-haiku-20240307") -> Dict:
    url = "https://api.anthropic.com/v1/messages"
    headers = {
        "x-api-key": os.getenv("ANTHROPIC_API_KEY"),
        "anthropic-version": "2023-06-01",
        "content-type": "application/json"
    }
    payload = {
        "model": model,
        "max_tokens": 4096,
        "messages": messages,
        "temperature": 0.1
    }
    response = requests.post(url, headers=headers, json=payload, timeout=30)
    
    # 关键：这里手动解析 rate limit 头
    if response.status_code == 429:
        remaining = response.headers.get('x-ratelimit-remaining', '0')
        reset = response.headers.get('x-ratelimit-reset', '0')
        raise RateLimitError(f"Rate limited. Remaining: {remaining}, Reset in {reset}s")
    
    response.raise_for_status()
    return response.json()

这个函数只有 22 行，但它把 x-ratelimit-remaining 这个关键信号暴露给了上层调度器，让重试策略有了决策依据。这就是“小而美”在 €40 预算下的力量——没有魔法，只有对每个字节的掌控。

4.2 SQLite 数据库 Schema 设计：一张表如何承载整个状态机

数据库只有一个核心表 task_runs ，结构极简但信息完备：

CREATE TABLE task_runs (
  id TEXT PRIMARY KEY,                    -- UUID v4, e.g. "run_abc123"
  graph_id TEXT NOT NULL,                 -- 对应 YAML 文件名, e.g. "contract_review"
  node_id TEXT NOT NULL,                  -- 节点 ID, e.g. "parse_pdf"
  status TEXT NOT NULL CHECK(status IN ('PENDING', 'RUNNING', 'SUCCESS', 'FAILED', 'PARTIAL_FAILURE', 'SKIPPED')),
  input_data TEXT,                        -- JSON string of input (truncated if > 4KB)
  output_data TEXT,                       -- JSON string of output (truncated if > 4KB)
  error_message TEXT,                     -- 仅当 status IN ('FAILED', 'PARTIAL_FAILURE') 时有值
  created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
  updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
  last_retry_at TIMESTAMP,                -- 上次重试时间，用于 backoff 计算
  retry_count INTEGER DEFAULT 0,
  sla_deadline TIMESTAMP,                 -- HumanGate 的 SLA 截止时间
  human_response TEXT                     -- HumanGate 的最终回复 JSON
);

看到没？没有冗余字段，没有外键，没有索引（除了主键）。但 status 字段的 6 种取值，已经覆盖了所有可能的状态流转。 input_data 和 output_data 虽然存为 TEXT，但我在应用层强制用 json.dumps(obj, ensure_ascii=False, separators=(',', ':')) 序列化，确保体积最小。 error_message 不存完整 traceback，只存业务错误码（如 "RATE_LIMIT_EXCEEDED" ）和一句话描述（如 "Anthropic API returned 429" ），因为 traceback 对故障排查帮助不大，反而占空间。

注意：SQLite 的 CURRENT_TIMESTAMP 是本地时间，不是 UTC。这在单机部署下完全没问题，因为所有时间比较都在同一时区完成。如果你要跨时区部署，才需要换成 datetime('now') 并显式处理时区。€40 预算下，别给自己找麻烦。

4.3 核心调度循环：一个永不退出的 while True 如何保证高可靠

调度器的主循环，是一个极其朴素的 while True ，但它被精心包裹了三层防护：

def main_loop():
    # 第一层：进程级守护
    signal.signal(signal.SIGTERM, lambda s, f: sys.exit(0))
    
    while True:
        try:
            # 第二层：单次循环超时保护
            run_once_with_timeout(timeout=60)  # 超过 60 秒强制中断
            
        except Exception as e:
            # 第三层：全局异常兜底
            logger.critical(f"Main loop crashed: {e}", exc_info=True)
            time.sleep(5)  # 等 5 秒再重启，避免疯狂打日志
            continue

def run_once_with_timeout(timeout: int):
    # 使用 signal.alarm 实现硬超时（Linux only）
    def timeout_handler(signum, frame):
        raise TimeoutError("Loop iteration timed out")
    
    signal.signal(signal.SIGALRM, timeout_handler)
    signal.alarm(timeout)
    
    try:
        # 执行真正的调度逻辑
        _execute_scheduling_cycle()
    finally:
        signal.alarm(0)  # 取消 alarm

_execute_scheduling_cycle() 函数做了四件事：

1. 扫描待执行节点 ： SELECT * FROM task_runs WHERE status = 'PENDING' AND (node_id NOT IN (SELECT node_id FROM task_runs WHERE status = 'RUNNING')) —— 确保同一节点不会并发执行。
2. 检查依赖就绪 ：对每个待执行节点，解析其 dependencies 字段（如 parse_pdf.raw_text ），然后查 SQLite 确认上游节点 status == 'SUCCESS' 且 output_data 中存在该字段且类型匹配。
3. 执行节点逻辑 ：调用对应的 call_claude() 或 call_codex() ，捕获所有异常并分类处理。
4. 更新状态 ：无论成功失败，都 UPDATE task_runs SET status = ?, output_data = ?, error_message = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ? 。

这个循环每 3 秒运行一次，但每次只处理最多 5 个任务（防止单次循环过长）。实测下来，一台 T14 在满负载时 CPU 占用稳定在 12%，内存 380MB，完全游刃有余。它的可靠性不来自复杂算法，而来自 每一次迭代的确定性、每一次更新的原子性、每一次异常的明确归因 。

4.4 Slack HumanGate 集成：50 行代码搞定企业级人机协同

Slack 集成不需要 OAuth 复杂流程。我用的是最简单的 Incoming Webhook + Slash Command 组合：

1. Incoming Webhook ：在 Slack App 后台创建，获得一个 https://hooks.slack.com/services/T00000000/B00000000/XXXXXXXXXXXXXXXXXXXXXXXX URL。 HumanGate 节点发消息时，只 POST 一个极简 payload：

{
  "channel": "#contract-review",
  "text": "New clause for review (Run ID: run_abc123)",
  "blocks": [
    {
      "type": "section",
      "text": {
        "type": "mrkdwn",
        "text": "*Payment Term on Page 7*\n\nAmount: USD 50,000\nDue Date: 30 days after delivery\nCurrency: USD"
      }
    },
    {
      "type": "actions",
      "elements": [
        {
          "type": "button",
          "text": {"type": "plain_text", "text": "Approve"},
          "value": "approve_run_abc123",
          "action_id": "approve"
        },
        {
          "type": "button",
          "text": {"type": "plain_text", "text": "Request Changes"},
          "value": "changes_run_abc123",
          "action_id": "changes"
        }
      ]
    }
  ]
}

2. Slash Command /resolve ：指向我的 Flask Webhook。处理函数只有 50 行：

@app.route('/slack/resolve', methods=['POST'])
def resolve_slash():
    # 1. 验证 Slack 签名
    if not verify_slack_signature(request):
        return "Forbidden", 403
    
    # 2. 解析 form data
    text = request.form.get('text', '').strip()
    channel_id = request.form.get('channel_id')
    user_id = request.form.get('user_id')
    
    # 3. 提取 run_id 和 action
    match = re.match(r'^(\w+)_(\w+)$', text)
    if not match:
        return "Invalid format. Use: /resolve approve_run_abc123", 200
    
    action, run_id = match.groups()
    
    # 4. 更新 SQLite
    conn = sqlite3.connect('/var/data/agent.db')
    cursor = conn.cursor()
    cursor.execute(
        "UPDATE task_runs SET status = ?, human_response = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?",
        ("RESOLVED", json.dumps({"action": action, "by": user_id}), run_id)
    )
    conn.commit()
    conn.close()
    
    return f"✅ Resolved '{action}' for {run_id}", 200

整个集成，没有第三方 SDK，没有 OAuth token 刷新，没有复杂的事件订阅。它就是一个 HTTP 接口，输入是 Slack 的 form data，输出是纯文本响应。简单，就是 €40 预算下最高的可靠性。

5. 常见问题与排查技巧实录：那些文档里绝不会写的坑

5.1 问题速查表：从现象到根因的 5 分钟定位法

现象	可能根因	快速验证命令	解决方案
parse_pdf 节点反复 PARTIAL_FAILURE ， output_data 里只有 {page_count: 12}	PDF 文本提取失败（加密/PDF/A 格式）	pdfinfo your_file.pdf | grep -i "encrypted|conformance"	用 qpdf --decrypt 尝试解密；或改用 pdfplumber 替代 pypdf
extract_clauses 节点卡在 RUNNING 状态超过 5 分钟	Codex endpoint 响应超时，且重试策略未生效	sqlite3 /var/data/agent.db "SELECT * FROM task_runs WHERE node_id='extract_clauses' AND status='RUNNING';"	检查 retry_count 是否为 0；确认 codex_endpoint 环境变量是否拼写错误
Slack 收不到任何消息，但日志显示 Webhook sent	Slack Webhook URL 过期或权限不足	curl -X POST -H 'Content-type: application/json' --data '{"text":"test"}' YOUR_WEBHOOK_URL	重新生成 Webhook URL；确认 Slack App 已安装到目标 workspace
HumanGate 的 /resolve 返回 403	Slack 签名验证失败（时钟不同步）	date 对比服务器和 Slack 服务器时间	sudo ntpdate -s time.nist.gov 同步时间；或在 verify_slack_signature 函数里放宽 5 分钟容忍窗口
调度器进程 CPU 100% 持续 10 分钟以上	run_once_with_timeout 的 signal.alarm 在某些 Python 版本下失效	ps aux | grep agent-scheduler 查看进程树	改用 threading.Timer 替代 signal.alarm ；或升级到 Python 3.11.6+

这张表是我踩了 17 次坑后总结的。它不教你理论，只告诉你“看到什么，立刻做什么”。比如，当你发现 parse_pdf 总是 PARTIAL_FAILURE ，第一反应不该是重写整个 PDF 解析模块，而是先跑 pdfinfo 看看文件是不是加密的——这一步通常 10 秒内就能定位 70% 的同类问题。

5.2 一个真实的故障复盘：Claude 的 max_tokens 陷阱如何烧掉 €2.3

上周五下午，系统突然开始大量 FAILED ，错误日志全是 {"error": {"type": "overload_error", "message": "Request was too large"}} 。我以为是流量突增，查了监控，QPS 没变。最后发现，是某份新合同的 PDF 里嵌入了 12 张高清扫描图， pypdf 提取文本时，把图片的二进制数据也当作文本读了出来，导致 raw_text 字符串长达 2.3MB。Claude 的 max_tokens: 4096 设置，在面对这种畸形输入时，根本不是限制输出长度，而是触发了服务端的 overload 保护。

解决方案分三步：

1. 前端过滤 ：在 parse_pdf 节点里加一行 if len(raw_text) > 500_000: raise ValueError("Text too long: {} chars".format(len(raw_text))) ，直接拒绝处理。
2. 降级路径 ：当触发此异常，自动启动 ocr_fallback 子流程：用 pytesseract 对 PDF 每页做 OCR，OCR 结果再喂给 Claude。虽然慢 3 倍，但保证了可靠性。
3. 告警升级 ：在调度器里加一条规则：如果 1 小时内 parse_pdf 的 FAILED 率 > 5%，则向我的 Telegram 发送告警，并附上最近 3 个失败的 pdf_url 。

这个故障让我明白： 可靠性不是靠堆砌防御，而是靠对每一个环节“最坏情况”的坦诚面对 。€40 预算买不来容错，只能买来直面真相的勇气。

5.3 给新手的三条血泪经验

1. 永远不要相信模型的“正常”输出 。Claude 的 {"error": ...} 和 Codex 的 {"error": ...} 结构完全不同；Slack 的 response_url 有时会返回 404（当用户删除了原始消息）；SQLite 的 INSERT OR REPLACE 在某些情况下会静默失败。我的做法是：对每一个外部交互，都写一个 validate_xxx_response() 函数，哪怕只是 assert isinstance(resp, dict) and 'content' in resp 。这多花的 30 秒，能省下你 3 小时 debug 时间。

2. 把“人类”当作一个有 SLA 的 API 来设计 。不要说“等审核员回复”，要说“等审核员在 30 分钟内回复，否则自动升级”。把人类的不确定性，转化为一个可测量、可补偿的业务指标。这是我从运维 SRE 那里偷来的思路。

3. €40 预算的终极奥义，是“不做选择题” 。不要纠结“该用 LangChain 还是 LlamaIndex”，因为答案是“都不用”；不要纠结“该用 Redis 还是 PostgreSQL”，因为答案是“用 SQLite”；不要纠结“该用 Claude 还是 Codex”，因为答案是“各司其职”。预算有限时，最大的生产力提升，来自于果断砍掉所有“看起来不错但非必要”的选项。我删掉了最初设计的“邮件通知”、“Telegram 告警”、“Grafana 监控面板”，只留下最核心的 SQLite 日志和一个 tail -f /var/log/agent.log 。结果？系统更稳了，因为干扰项少了。

我在实际部署中发现，最常被忽略的其实是日志的“可操作性”。很多日志只写 Task failed ，而我的日志一定包含 Task 'parse_pdf' failed with RATE_LIMIT_EXCEEDED at 2024-04-15T14:22:03Z. Run ID: run_abc123. Check ANTHROPIC_API_KEY quota. —— 这样，你不用打开任何其他工具，光看日志就能知道下一步该做什么。这才是 €40 预算下，真正的可靠性。