#!/usr/bin/env python3 """02_workflow_vs_agent_decision.py — workflow vs agent 决策矩阵演示（无需 API key） 把官方「When (and when not) to use agents」的决策阶梯编码为确定性规则引擎： 1. 先找最简单的方案（单次 LLM 调用 + 检索 + in-context examples 往往就够） 2. 复杂度确需提升时：well-defined 任务 -> workflow（预测性/一致性） 3. 规模上需要灵活性与模型驱动决策 -> agent（开放式、步数不可预测） 再按任务特征细选编排模式：routing / prompt chaining / parallelization / evaluator-optimizer / orchestrator-workers / agent / multi-agent。 官方依据（访问于 2026-08-30）: https://www.anthropic.com/engineering/building-effective-agents https://www.anthropic.com/engineering/multi-agent-research-system 运行: python3 02_workflow_vs_agent_decision.py """ from __future__ import annotations from dataclasses import dataclass @dataclass(frozen=True) class Scenario: """场景特征向量 —— 决策只看这些属性，不看行业词汇。""" name: str steps_predictable: bool # 步骤能否预先枚举（官方: task can be cleanly decomposed） distinct_categories: bool # 输入是否存在清晰类别（routing 信号） independent_subtasks: bool # 子任务是否彼此独立（parallelization / workers 信号） subtask_set_varies: bool # 子任务集合是否随输入变化（orchestrator-workers 信号："subtasks aren't pre-defined, but determined by the orchestrator"） clear_eval_criteria: bool # 是否有清晰评估标准（evaluator-optimizer 信号） fixed_sequence: bool # 子任务是否固定顺序依赖（chaining 信号） exceeds_single_context: bool # 信息量是否超单上下文窗口（multi-agent 信号） high_value_task: bool # 任务价值是否高到能兜住 ~15x token（multi-agent 前提） single_call_enough: bool # 一次调用 + 检索/示例是否已够（最简方案优先） def decide(s: Scenario) -> tuple[str, str, list[str]]: """返回 (推荐架构, 核心理由, 风险提示)。规则顺序 = 官方决策阶梯的顺序。""" if s.single_call_enough: return ( "single augmented LLM call", "官方第一原则：先找最简单的可行方案；很多应用单次调用+检索+示例就够", ["若未来复杂度上升，再沿阶梯升级，不要预支复杂度"], ) if not s.steps_predictable and s.independent_subtasks and s.exceeds_single_context and s.high_value_task: return ( "multi-agent (orchestrator-workers)", "开放式问题 + 可高度并行 + 信息超单窗口 + 高价值兜住 ~15x token", ["强依赖/需共享上下文的子任务不适合", "委派指令必须含目标/输出格式/工具指引/边界"], ) if not s.steps_predictable: return ( "agent (single, with stopping conditions)", "步数与路径无法预先枚举，需要模型运行时决策；官方提醒这是拿延迟和成本换任务表现", ["必须配 max iterations / 预算上限 / 人审检查点；sandbox 大量试跑"], ) # steps_predictable == True 从这里往下全部是 workflow 家族 if s.distinct_categories: return ( "workflow: routing", "输入类别清晰，分离关注点；各类走专用 prompt/工具，甚至专用（更便宜的）模型", ["分类器准确率决定上限；新增类别要更新路由"], ) if s.fixed_sequence: return ( "workflow: prompt chaining", "固定顺序子任务；用延迟换准确率，每步更简单，可加程序化 gate", ["串行延迟=各步之和；gate 过严会频繁中断"], ) if s.clear_eval_criteria: return ( "workflow: evaluator-optimizer", "有清晰评估标准且迭代精修有可测收益；生成器+评估器循环", ["无清晰标准时评估器会漂移；循环要设上限"], ) if s.subtask_set_varies: return ( "workflow: orchestrator-workers", "步骤可枚举但子任务集合随输入变化，由编排器现场分解并综合（BEA 官方例：多文件编码改动）", ["比纯并行化灵活但可预测性下降一档；worker 委派单要含目标/输出格式/边界"], ) if s.independent_subtasks: return ( "workflow: parallelization (sectioning/voting)", "子任务独立可并行（或需多视角投票），耗时≈最慢子任务", ["聚合逻辑要程序化确定性实现"], ) return ( "workflow: prompt chaining (fallback)", "无独立子任务/类别/评估标准且集合固定：退化为顺序链即可满足", ["若实测发现子任务集合其实随输入变化，升级为 orchestrator-workers"], ) SCENARIOS = [ Scenario("内部文档问答（RAG 即答）", subtask_set_varies=False, steps_predictable=True, distinct_categories=False, independent_subtasks=False, clear_eval_criteria=False, fixed_sequence=False, exceeds_single_context=False, high_value_task=False, single_call_enough=True), Scenario("客服全量分流（咨询/退款/技术支持）", subtask_set_varies=False, steps_predictable=True, distinct_categories=True, independent_subtasks=False, clear_eval_criteria=False, fixed_sequence=False, exceeds_single_context=False, high_value_task=False, single_call_enough=False), Scenario("营销文案生成后翻译", subtask_set_varies=False, steps_predictable=True, distinct_categories=False, independent_subtasks=False, clear_eval_criteria=False, fixed_sequence=True, exceeds_single_context=False, high_value_task=False, single_call_enough=False), Scenario("代码漏洞审查（多 prompt 各审一遍再投票）", subtask_set_varies=False, steps_predictable=True, distinct_categories=False, independent_subtasks=True, clear_eval_criteria=False, fixed_sequence=False, exceeds_single_context=False, high_value_task=False, single_call_enough=False), Scenario("文学翻译精修（有明确评审标准）", subtask_set_varies=False, steps_predictable=True, distinct_categories=False, independent_subtasks=False, clear_eval_criteria=True, fixed_sequence=False, exceeds_single_context=False, high_value_task=False, single_call_enough=False), Scenario("跨 50 信源行业研究报告", subtask_set_varies=True, steps_predictable=False, distinct_categories=False, independent_subtasks=True, clear_eval_criteria=False, fixed_sequence=False, exceeds_single_context=True, high_value_task=True, single_call_enough=False), Scenario("修一个未知根因的生产 bug 并部署", subtask_set_varies=False, steps_predictable=False, distinct_categories=False, independent_subtasks=False, clear_eval_criteria=False, fixed_sequence=False, exceeds_single_context=False, high_value_task=False, single_call_enough=False), Scenario("按需求改多文件的重构（子任务集合随需求变化）", subtask_set_varies=True, steps_predictable=True, distinct_categories=False, independent_subtasks=True, clear_eval_criteria=False, fixed_sequence=False, exceeds_single_context=False, high_value_task=False, single_call_enough=False), ] EXPECTED = [ "single augmented LLM call", "workflow: routing", "workflow: prompt chaining", "workflow: parallelization (sectioning/voting)", "workflow: evaluator-optimizer", "multi-agent (orchestrator-workers)", "agent (single, with stopping conditions)", "workflow: orchestrator-workers", ] def main() -> None: header = f"{'场景':<28} {'推荐架构':<46} 核心理由" print(header) print("-" * 132) failures = 0 for s, expected in zip(SCENARIOS, EXPECTED): arch, why, risks = decide(s) ok = arch == expected failures += 0 if ok else 1 mark = "OK " if ok else "!! " print(f"{mark}{s.name:<25} {arch:<44} {why}") for r in risks: print(f" risk: {r}") print("-" * 132) if failures: raise SystemExit(f"{failures} 个场景与预期不符，请检查规则引擎") print("8 个场景全部与官方决策阶梯一致：先最简方案，well-defined 用 workflow，" "开放式才上 agent，开放式+可并行+超窗口+高价值才上 multi-agent。") if __name__ == "__main__": main()