实用提示:众人拾柴,一支笔:在协调多个代理的同时不丢失……
《实用笔记:众人拾柴,一支笔——在协调各代理的同时保留合同、校验项以及适用于采用该模式的团队的即插即用代码模块》的操作指南。
可将此内容视为《众手一支笔:在掌控全局的同时协调智能体》中理念的面向操作员的简化版本:清晰的阶段划分、有序的代码模块,以及能在交接过程中保留的恢复说明。将“概览”阶段视为可度量的基准最为有效,在扩大范围之前,需记录一份最佳案例、一个故障实例以及对应的回滚说明。相比庞大的脚本,应优先使用小型且可测试的单元。当某一步骤出现故障时,故障原因应能明确指向某个具体责任方,而非复杂的流程链。
071 · 采用工作流机制,让智能体赢得自主权
对于071阶段,默认情况下,在修改代码之前需先定义输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤,而无需猜测隐藏状态。 应将此阶段视为输入与已验证输出之间的契约。为相关成果命名,明确成功判定标准,并拒绝默许的半完成状态。 对于涉及资金支出或修改生产数据的操作,必须经过人工审批。编译时的连接方式并不等同于业务上的完整性。
REFUND_LIMIT = 50.00
def llm_step(prompt: str) -> str:
"""Called only where the written rules run out."""
return model.complete(prompt).strip().lower()
def handle_refund(ticket):
if ticket.days_since_purchase > 30:
return deny(ticket, reason="outside window")
if ticket.amount <= REFUND_LIMIT:
return approve(ticket) # a rule, not a judgement call
intent = llm_step(
f"Classify as fraud, defect, or remorse:\n{ticket.body}"
)
if intent == "fraud":
return escalate(ticket, queue="risk")
if intent == "defect":
return approve(ticket)
return route_to_human(ticket)
072 · 已知分解结构时可直接跳过编排器
在072“跳过编排器”阶段,应在修改代码之前明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤,而无需猜测隐藏状态。应在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况,可避免在从演示环境过渡到共享环境时出现意外费用。对于会消耗资金或修改生产数据的操作,必须经过人工审批。编译时的连接方式并不等同于业务功能的完整性。
# Before: up to 15 planning calls to rediscover a fixed list
def run_orchestrated(doc):
state = {"doc": doc}
for _ in range(15):
decision = orchestrator.plan(state) # 1 LLM call per pass
if decision.action == "finalize":
break
state = WORKERS[decision.worker](state)
return state
# After: 0 planning calls, same four workers, same result
PIPELINE = [fetch, extract, summarize, format_report]
def run_static(doc):
state = {"doc": doc}
for step in PIPELINE: # 0 LLM calls here
state = step(state)
return state
073 · 按上下文边界而非职位划分代理
对于按阶段划分的073 Split Agents,修改代码之前需明确输入参数、该步骤的负责人以及结束条件。操作人员应能够从已知的检查点重新运行该步骤,而无需猜测隐藏状态。 配置信息应置于应用程序代码之外。环境文件、密钥存储以及功能标志应集中存放于一个位置,以便操作人员无需查看整个流程即可进行审计。 对于涉及资金支出或修改生产数据的节点,需设置人工审批环节。编译时的连接方式并不等同于业务流程的完整性。 对于按阶段划分的073 Split Agents,修改代码之前需明确输入参数、该步骤的负责人以及结束条件。操作人员应能够从已知的检查点重新运行该步骤,而无需猜测隐藏状态。 相较于庞大的脚本,应优先使用小型且可测试的单元。当某个步骤失败时,故障原因应能明确指向单一责任模块,而非复杂的流程链。
from dataclasses import dataclass
@dataclass
class Subtask:
name: str
needs: set[str] # facts this subtask must read
produces: set[str] # facts this subtask decides
def should_split(a: Subtask, b: Subtask) -> bool:
"""Split only when neither side needs what the other decides."""
shared = (a.needs & b.produces) | (b.needs & a.produces)
return not shared
implement = Subtask("implement", {"spec"}, {"api_shape", "error_semantics"})
test = Subtask("test", {"spec", "api_shape", "error_semantics"}, {"cases"})
assert not should_split(implement, test) # one agent writes both
074 · 调度器应直接给出决策,而非调用工具
在处理074“调度器应”这一阶段时,首先需明确相关契约:所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 将这一阶段视为输入与验证后输出之间的契约。为相关产物命名,定义成功判定标准,并杜绝无声的半完成状态。 需记录每次调用的工具名称、参数哈希值、延迟时间以及最终结果。若没有这些记录,调试代理将陷入无止境的循环,耗费大量时间。
from typing import Literal
from pydantic import BaseModel
class OrchestratorDecision(BaseModel):
next_action: Literal["delegate", "replan", "finalize"]
target_worker: str | None = None
task_description: str | None = None
reasoning: str
def orchestrator_step(state):
d = decide(state) # model has zero tools attached
if d.next_action == "finalize":
return finalize(state, d.reasoning)
if d.next_action == "replan":
return state.reset_plan(d.reasoning)
worker = WORKERS[d.target_worker] # workers own every tool
result = worker.run(d.task_description)
return state.record(d.target_worker, result)
075 · 为每个子代理提供类型化的输出契约,而非自由文本
在处理075“为每个子代理提供资源”阶段时,首先写下相关契约:所需输入、成功标志以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 在功能结果旁记录执行时间以及令牌或查询成本。提前明确成本有助于避免从演示环境过渡到共享环境时出现意外费用。 在耗时较高的步骤之后设置检查点。当操作员重新尝试后续节点时,恢复流程不应再次收取相同的LLM调用费用。
from pydantic import BaseModel, Field, ValidationError
class SubagentResult(BaseModel):
findings: list[str] = Field(max_length=5) # hard cap, one sentence each
sources: list[str]
open_questions: list[str] = []
completion_status: str # complete | partial | blocked
def parse_or_retry(worker, task, attempts=2):
for _ in range(attempts):
raw = worker.run(task, response_format=SubagentResult)
try:
return SubagentResult.model_validate_json(raw)
except ValidationError as err:
task = f"{task}\n\nRejected: {err}\nReturn only JSON in the schema."
raise RuntimeError(f"{worker.name} returned no valid result")
076 · 通过一个代理实现并行读取与集中写入
在处理076 Fan Out Reads阶段时,首先需明确相关规范:所需的输入参数、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 应将配置信息与应用程序代码分开存放。环境文件、密钥存储以及功能开关应集中于一个位置,以便操作员无需查看整个流程即可进行审计。 在耗时较高的步骤之后设置检查点。当操作员重新尝试后续节点时,恢复流程不应再次调用相同的LLM接口。 在处理076 Fan Out Reads阶段时,首先需明确相关规范:所需的输入参数、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 相较于庞大的脚本,更应优先使用小型且可测试的单元。当某个步骤失败时,故障应能指向具体的责任模块,而非整个复杂的流程。
import asyncio
READ_TOOLS = ["search_repo", "read_file", "fetch_docs"]
async def gather_context(subtasks):
workers = [Agent(name=t.name, tools=READ_TOOLS) for t in subtasks]
return await asyncio.gather(
*(w.run(t.prompt) for w, t in zip(workers, subtasks))
)
async def build_feature(spec, subtasks):
findings = await gather_context(subtasks) # wide, parallel, read-only
writer = Agent(name="writer", tools=["write_file", "apply_patch"])
return await writer.run(spec, context=findings) # one writer, one pass
077 · 记录终止条件:开发人员确实不知道何时该停止
将077阶段视为可度量的工作面时效果最佳。在扩大范围之前,先收集一份优秀的代码示例、一个失败案例以及回滚说明。 把这一阶段视为输入与经过验证的输出之间的契约。为相关文档命名,明确成功标准,杜绝默许的半完成状态。 保持数据结构扁平且类型明确。嵌套的数据块会掩盖哪个节点编写了哪个字段,还会在中断后导致程序无法继续运行。
CHECKLIST = [
"every requested section exists",
"each claim cites a retrieved source",
"open questions are listed, or explicitly none",
]
def verify_complete(objective, checklist, output) -> bool:
for item in checklist:
verdict = judge(f"Objective: {objective}\nCheck: {item}\n\n{output}")
if not verdict.passed:
log.info("termination blocked by: %s", item)
return False
return judge(f"Does this satisfy the objective?\n{objective}\n\n{output}").passed
def finish(state):
if verify_complete(state.objective, CHECKLIST, state.draft):
return state.done()
return state.keep_working(reason="checklist not satisfied")
操作检查清单
在处理操作检查清单阶段时,首先要写下契约内容:所需输入、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。
应将正常流程和恢复流程一并记录下来。重试机制、人工干预环节以及死信处理都是产品本身的组成部分,而非后续才添加的完善措施。
在成本较高的操作之后设置检查点。当操作员重新尝试后续节点时,恢复功能不应再次计费相同的LLM调用费用。
锁定依赖版本的编号,并记录用于运行演示的镜像摘要。可重复性远比团队内部的经验知识更重要。
优先选择小型且易于测试的单元,而非结构复杂的脚本。当某个步骤失败时,故障应能指向具体的责任模块,而非整个混乱的流程。
在成本较高的操作之后设置检查点。当操作员重新尝试后续节点时,恢复功能不应再次计费相同的LLM调用费用。
在推广该技术栈之前,应先冻结版本、为关键流程记录标准输出日志,并明确回滚步骤。共享环境需要设置速率限制、租户验证机制,以及明确的密钥轮换负责人。与其展示花哨的一次性演示,不如注重扎实的可靠性。
针对 64be605517f1 的批量处理说明:请将提供商密钥移出代码仓库,设定单会话令牌上限,并将日志存储在评估用示例文件旁,以便后续模型更换时保持数据可比性。
对于强化安全性的第 0 阶段,在修改代码之前需明确输入参数、该步骤的负责人以及完成标准。操作人员应能够从已知的检查点重新运行该步骤,而无需猜测隐藏状态。需同时记录正常流程与故障恢复流程。重试机制、人工审核环节以及错误处理措施都是产品不可或缺的部分,而非后续才需要补充的功能。
强化细节 0/768:记录该代码段的运行时间、错误类型以及代币消耗情况,然后依据固定的评估标准而非个人经验来决定是否保留该修改。
在处理强化步骤1时,首先写下合约的详细内容:所需的输入参数、成功标志,以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。将此阶段视为输入与验证后输出之间的契约,为相关组件命名、明确成功判定标准,并杜绝无声的半完成状态。
强化细节 1/768:记录该代码段的运行时间、错误类型以及代币消耗情况,然后依据固定的评估标准而非个人经验来决定是否保留该修改。
将强化措施的第2阶段视为可测量的表面时效果最佳。在扩大范围之前,先记录一份理想的运行日志、一个故障案例以及回滚说明。 应将配置与应用程序代码分开。环境文件、密钥存储和功能标志应集中存放于一处,这样操作人员无需查看整个系统结构即可进行审计。
强化措施细节2/768:针对此条措施需测量耗时、错误类型以及令牌使用情况,然后依据固定的问题清单而非个人经验来判断是否保留该变更。
在实施强化措施的第3阶段,应在修改代码之前明确输入参数、该步骤的负责人以及完成标准。操作人员应能够从已知的检查点重新运行该步骤,而无需猜测隐藏状态。相比冗长的脚本,更应采用小型且可测试的单元。当某一步骤失败时,故障原因应能指向具体的责任主体,而非复杂的流程问题。
强化措施细节3/768:需记录该步骤的运行时间、错误类型以及令牌消耗情况,然后依据固定的评估标准而非主观判断来决定是否保留该变更。
在处理强化措施的第4阶段时,首先写下相关约定:所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改始终符合初始要求。
在功能结果旁记录执行时间以及令牌或查询成本。提前了解这些成本信息,可避免在系统从演示环境过渡到共享环境时出现意外费用。
强化措施细节4/768:为该阶段测量实际执行时间、错误类型以及令牌消耗情况,然后依据固定的评估标准而非个人经验来决定是否保留相关修改。
将强化措施的第5阶段视为可测量的对象来处理效果最佳。在扩大范围之前,先记录一个理想运行案例、一个失败案例以及回滚说明。
应同时记录正常流程和故障恢复流程。重试机制、人工审核环节以及错误处理方式都是产品本身的组成部分,而非后续需要补充的内容。
强化措施细节5/768:记录该任务的执行时间、错误类型以及令牌消耗情况,然后依据固定的评估标准而非个人经验来决定是否保留该变更。
在强化措施的第6阶段,应在修改代码之前明确输入参数、该步骤的负责人以及完成标准。操作人员应能够从已知的检查点重新运行该步骤,而无需猜测隐藏状态。应将此阶段视为输入参数与验证后输出结果之间的契约,为相关成果命名、定义成功判定条件,并拒绝默许的半完成状态。
强化措施细节6/768:记录该任务的执行时间、错误类型以及令牌消耗情况,然后依据固定的评估标准而非个人经验来决定是否保留该变更。
在处理强化措施的第7阶段时,首先写下相关契约:所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改始终符合要求。 应将配置信息与应用程序代码分开。环境文件、密钥存储以及功能开关应集中存放于一个位置,这样操作人员无需查看整个系统结构即可进行审计。
强化措施细节7/768:针对该措施需统计执行耗时、错误类型以及令牌使用情况,然后依据固定的评估标准而非个人经验来决定是否保留该变更。