依赖感知的智能体编排：用SQLite和Claude/Codex构建高可靠AI工作流

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”的架构。

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

注意看最后一行：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：审核员通过预设的/resolveSlash 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 的真实行为（比如openaiSDK 会自动重试 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/XXXXXXXXXXXXXXXXXXXXXXXXURL。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 分钟定位法

这张表是我踩了 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 预算下，真正的可靠性。