实用笔记：修复AI智能体循环中的非确定性回放错误

以下笔记为“解决AI智能体循环中的非确定性重放错误”提供了一条实用路径。重点在于契约、校验以及可直接插入的代码占位符，而非动机性阐述。 在完成概览阶段时，首先写下契约内容：所需输入、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。

错误本身，先于名称

在扩展范围之前，应将阶段前的缺陷视为可测量的对象来处理。先收集一份成功的示例、一个故障案例以及回滚说明。 将配置与应用程序代码分开。环境文件、密钥存储和功能标志应集中存放，以便操作人员无需查看整个系统结构即可进行审计。 保持系统状态的扁平化与类型化。嵌套的数据结构会掩盖是哪个节点修改了哪个字段，还会导致在中断后无法继续执行。

# ❌ What I almost wrote — LLM call INSIDE the workflow.
@workflow.defn
class SourcingWorkflow:
    @workflow.run
    async def run(self, brief: SourcingBriefInput) -> SourcingResult:
        suppliers = await workflow.execute_activity(research_activity, brief)
        scored = await workflow.execute_activity(score_activity, suppliers)

        # The LLM call — right here, in the workflow. THIS IS THE BUG.
        # If worker dies here, replay will re-call the LLM and mutate state/history.
        response = await llm_client.complete(
            messages=build_decide_prompt(scored),
            temperature=0.0,
        )
        selected = parse_llm_decision(response.content, scored)

        approval = await workflow.wait_condition(...)
        # ...create PO, initiate payment

# ✅ The fix — LLM call moved into an activity.
@activity.defn
async def decide_activity(scored: list[dict], brief: SourcingBriefInput) -> dict:
    """The LLM call lives here — in the activity, not the workflow."""
    from app.agentmesh.llm import get_llm_client

    client = get_llm_client()
    response = await client.complete(
        messages=build_decide_prompt(scored),
        temperature=0.0,
    )
    selected, rationale = parse_llm_decision(response.content, scored)
    return {"selected_supplier": selected, "decision_reason": rationale}


@workflow.defn
class SourcingWorkflow:
    @workflow.run
    async def run(self, brief: SourcingBriefInput) -> SourcingResult:
        suppliers = await workflow.execute_activity(research_activity, brief)
        scored = await workflow.execute_activity(score_activity, suppliers)

        # The LLM call is NOWHERE in the workflow.
        # The activity result is recorded. On replay, it's injected.
        decision = await workflow.execute_activity(
            decide_activity,
            args=(scored, brief),
        )

原则：重放操作必须产生相同的历史记录

将法律重放过程视为可测量的对象时，其效果会最佳。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。同时记录正常流程与恢复流程的细节。重试机制、人工审核环节以及死信处理都是产品本身的一部分，而非后续补充的功能。要保持图表状态简洁且具有类型定义，嵌套的数据结构会掩盖具体是哪个节点修改了哪个字段，还会在流程中断后导致无法继续执行。

违规情况：当大语言模型被纳入工作流时会发生什么

在“违规行为发生阶段”的处理中，将其视为可测量的对象会更为有效。在扩大范围之前，先记录一份理想的处理结果、一个故障案例以及回滚说明。 相较于复杂的脚本，应优先选择小型且可测试的单元。当某个步骤出现故障时，故障应指向单一的责任主体，而非错综复杂的流程。 需为每轮对话和每次会话设定token预算。智能工具往往会大量消耗上下文资源，设置上限可避免演示过程变成意外的费用账单。 在“违规行为发生阶段”的处理中，将其视为可测量的对象会更为有效。在扩大范围之前，先记录一份理想的处理结果、一个故障案例以及回滚说明。 除了功能结果外，还需记录执行时间以及token或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外账单。

为何“temperature=0.0”无法帮你规避问题

对于“Why温度0 0阶段”，在修改代码之前需明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 配置信息应置于应用程序代码之外。环境文件、密钥存储以及功能标志应集中存放于一个位置，以便操作人员无需查看整个流程即可进行审计。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的连接方式并不等同于业务流程的完整性。

边界：活动即非确定性屏障

在更改代码之前，需为活动阶段界定输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 需同时记录正常流程与异常恢复路径。重试机制、人工审核环节以及错误处理都是产品本身的组成部分，而非后续需要补充的内容。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。仅靠编译时的配置并不足以确保业务的完整性。

代码：AgentMesh中的实际实现形式

在修改代码之前，需明确该阶段的输入参数、负责执行该步骤的人员以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测其中的隐藏状态。相比冗长的脚本，更应采用小型且易于测试的单元。当某个步骤失败时，故障原因应能明确指向具体的责任方，而非整个复杂的流程。对于涉及资金支出或修改生产数据的操作，必须经过人工审批。仅靠编译时的配置并不足以确保业务的完整性。在修改代码之前，需明确该阶段的输入参数、负责执行该步骤的人员以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测其中的隐藏状态。除了功能测试结果外，还需记录执行时间以及令牌或查询的成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外的费用支出。

Temporal Workflow (deterministic — replayed)
    └── Activity: run_graph_until_interrupt (non-deterministic — recorded once)
            └── LangGraph StateGraph
                    └── decide_node (async function)
                            └── get_llm_client().complete()  ← the LLM call

工作流——纯粹的编排机制（workflow.py）

在处理纯粹的编排阶段时，首先需明确相关规范：所需的输入参数、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改不会出错。 配置信息应与应用程序代码分开存放。环境文件、密钥存储以及功能开关都应集中管理，这样操作人员无需查看整个流程图即可进行审核。 在耗时较高的步骤之后设置检查点。当操作人员重新执行后续节点时，恢复流程不应再次调用相同的大型语言模型。

@workflow.defn
class SourcingWorkflow:
    @workflow.run
    async def run(self, brief: SourcingBriefInput) -> SourcingResult:
        # ↓ This is the boundary. The activity runs once. Result is recorded.
        graph_result = await workflow.execute_activity(run_graph_until_interrupt, brief, ...)

        # Wait for human signal — a Temporal primitive, not an LLM call.
        # On replay, the signal is injected from history.
        await workflow.wait_condition(lambda: self._approval_received, timeout=timedelta(hours=24))

        # ↓ Another boundary. Resume activity runs once. Result is recorded.
        resume_result = await workflow.execute_activity(resume_graph, self._approval_data, ...)

        # ↓ Side effects — each is its own activity with its own retry policy.
        po_result = await workflow.execute_activity(create_po_activity, args=(...), ...)

活动组件——非确定性所在之处（activity.py）

在处理相关任务时，首先写下契约：所需的输入、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 同时记录正常流程和异常恢复流程。重试机制、人工审核环节以及死信处理都是产品本身的组成部分，而非后续的优化工作。 在成本较高的步骤之后设置检查点。当操作员重新尝试某个节点时，恢复流程不应再次调用相同的大语言模型接口。

@activity.defn
async def run_graph_until_interrupt(brief: SourcingBriefInput) -> dict:
    graph = build_sourcing_graph(checkpointer=await get_checkpointer())
    config = {"configurable": {"thread_id": activity.info().workflow_id}}

    # ↓ Everything inside this call is non-deterministic. It runs ONCE.
    #   The result dict is recorded in the event history. On replay, it's injected.
    final_state = await graph.ainvoke({"brief": brief, "suppliers": [], "attempts": 0, ...}, config)

    return {"paused": True, "selected_supplier": selected, "suppliers": final_state.get("suppliers", [])}

图节点——大语言模型实际运行的位置（graph.py）

在处理“阶段节点”时，首先需明确相关契约：所需输入、成功信号以及部分失败时的处理方式。这份清单能确保后续的代码修改保持透明可溯。 相较于冗长的脚本，应优先选择小型且易于测试的单元。当某一步骤失败时，故障应指向单一责任点，而非复杂的流程链。 缓存稳定的系统指令和工具结构。重复发送相同的开头信息是造成资源浪费的常见原因。 在处理“阶段节点”时，首先需明确相关契约：所需输入、成功信号以及部分失败时的处理方式。这份清单能确保后续的代码修改保持透明可溯。 在功能结果旁记录执行时间以及代币或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外账单。

async def decide_node(state: AgentState) -> dict:
    scored = state.get("scored_suppliers", [])
    messages = build_decide_prompt(brief.item, brief.quantity, brief.budget, scored, past_decisions)

    # ↓ THE LLM CALL. This is the non-determinism that must never be in the workflow.
    response = await get_llm_client().complete(messages=messages, temperature=0.0, max_tokens=1000)

    selected, rationale = parse_llm_decision(response.content, scored)
    return {"selected_supplier": selected, "decision_reason": rationale, "cost_incurred": response.cost_usd}    # ↓ THE LLM CALL. This is the non-determinism that must never be in the workflow.
    response = await get_llm_client().complete(messages=messages, temperature=0.0, max_tokens=1000)

当边界变得模糊时

将边界视为可测量的表面时，其在阶段划分中的作用最为显著。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。 将配置与应用程序代码分开。环境文件、密钥存储和功能标志应集中存放于一处，这样操作人员无需查看整个结构即可进行审计。 保持图结构的扁平化与类型化。嵌套的数据块会掩盖哪个节点修改了哪个字段的信息，还会导致在中断后无法继续处理。

Temporal event history             LangGraph checkpoint (Postgres)
  └── ActivityCompleted(result)      └── graph state at interrupt()
        paused: True                       node: "approve"
        selected: SupplierB                 selected: SupplierB
        suppliers: [A, B, C]                suppliers: [A, B, C]

隔离模式——处处遵循相同约束

当将隔离模式视为可测量的界面时，其在同一阶段的表现最为理想。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。同时记录正常流程和恢复流程的文档。重试机制、人工审核环节以及死信处理都是产品本身的一部分，而非后续需要补充的内容。保持图结构的状态简洁且具有类型定义，嵌套的数据块会掩盖哪个节点编写了哪个字段的信息，还会在中断后导致无法继续执行。

其他代理架构中的相同限制

在阶段层面，同样的约束条件若被视为可度量的对象，则效果最佳。在扩大范围之前，先记录一份理想的执行日志、一个故障案例以及回滚说明。 相较于庞大的脚本，应优先选择小型且可测试的单元。当某一步骤出现故障时，故障原因应能明确指向某个具体责任模块，而非复杂的流程链。 保持图结构的状态简洁且具有类型定义。嵌套的数据块会掩盖哪个节点修改了哪个字段的信息，还会在流程中断后导致无法继续执行。 在阶段层面，同样的约束条件若被视为可度量的对象，则效果最佳。在扩大范围之前，先记录一份理想的执行日志、一个故障案例以及回滚说明。 除了功能结果外，还需记录执行时间以及令牌或查询成本。尽早了解这些成本信息，可避免在流程从演示环境转向共享环境时出现意外费用。

成本：为确定性所付出的代价

在修改代码之前，需明确相关成本、输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 配置信息应置于应用程序代码之外。环境文件、密钥存储以及功能开关应集中存放，以便操作人员无需查看整个流程即可进行审计。 对于涉及资金支出或修改生产数据的节点，必须设置人工审批环节。编译时的连接方式并不能保证业务流程的完整性。

边缘情况1：长时间运行的任务与心跳检测问题

对于边缘情况1中的长运行阶段，在修改代码之前需明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 需同时记录正常流程与异常恢复流程。重试机制、人工审核环节以及错误消息处理都是产品功能的一部分，而非后续需要补充的内容。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的配置并不等同于业务功能的完整性。

@activity.defn
async def run_graph_until_interrupt(brief: SourcingBriefInput) -> dict:
    # Long-running: graph may run 5+ minutes with multiple LLM calls
    for node in graph.stream(initial_state, config):
        activity.heartbeat()  # ← "I'm alive, don't timeout me"
        # ... process node output

边缘情况2：通过Temporal将大语言模型令牌流式传输回去

对于“边缘情况2流处理阶段”，在修改代码之前需明确输入参数、该步骤的负责人以及终止条件。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测其中的隐藏状态。 相较于冗长的脚本，更应采用小型且可测试的单元。当某个步骤失败时，故障原因应能明确指向某个具体职责，而非整个复杂的处理流程。 如果后续步骤是代码调用或工具调用，相比自由形式的文字描述，带有结构化格式及模式验证的输出更为合适。 对于“边缘情况2流处理阶段”，在修改代码之前需明确输入参数、该步骤的负责人以及终止条件。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测其中的隐藏状态。 除了功能结果外，还需记录处理时间以及令牌或查询成本。提前了解成本情况，可避免在流程从演示环境转向共享环境时出现意外费用。

边缘情况3：任务重试策略——并非所有任务都应采用相同的重试方式

在处理边缘情况3的任务阶段时，首先明确相关规范：所需输入、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 将配置信息置于应用程序代码之外。环境文件、密钥存储以及功能开关应集中存放，以便操作员无需查看整个系统结构即可进行审计。 在耗时较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次调用相同的大型语言模型。

# Side effect — strict, non-retryable. Idempotency key handles safety.
po_result = await workflow.execute_activity(
    create_po_activity,
    args=(supplier, item, quantity, price),
    retry_policy=workflow.RetryPolicy(
        initial_interval=timedelta(seconds=1),
        maximum_attempts=1,  # ← don't retry. Idempotency key prevents duplicates.
        non_retryable_error_types=["DuplicatePOError"],
    ),
)

# Verification — aggressive retry, but with backoff for eventual consistency
po_verification = await workflow.execute_activity(
    verify_po_exists,
    args=(po_result["po_id"],),
    retry_policy=workflow.RetryPolicy(
        initial_interval=timedelta(seconds=2),  # ← give the DB time to sync
        maximum_attempts=5,
    ),
)

思维模型

在处理“思维模型”阶段时，首先写下相关约定：所需的输入参数、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 同时记录正常流程和异常恢复流程。重试机制、人工审核环节以及错误消息处理都是产品本身的组成部分，而非后续需要补充的功能。 缓存系统中稳定的指令和工具结构。重复发送相同的开头信息是导致资源浪费的常见原因。

下一篇将介绍什么

在“即将处理的内容”阶段工作时，首先需明确相关规范：所需输入、成功标志以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 建议使用小型、可测试的单元，而非冗长的脚本。当某个步骤失败时，故障应指向单一责任模块，而非复杂的流程链。 在成本较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次调用相同的大型语言模型。 在“即将处理的内容”阶段工作时，首先需明确相关规范：所需输入、成功标志以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 在功能结果旁记录执行时间以及token或查询成本。提前了解成本情况，可避免从演示环境过渡到共享环境时出现意外收费。

参考资料

将“参考资料”阶段视为可度量的对象来处理效果最佳。在扩大范围之前，先记录一份优秀的示例、一个失败案例以及回滚说明。 将配置与应用程序代码分开。环境文件、密钥存储和功能标志应集中存放于一个位置，以便操作人员无需查看整个系统结构即可进行审计。 保持系统状态的扁平化与类型化。嵌套的数据结构会掩盖哪个节点修改了哪个字段的信息，且在中断后会导致无法继续处理。

操作检查清单

在“操作检查清单”阶段，应在修改代码之前明确输入内容、该步骤的负责人以及完成标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏的状态。 应将此阶段视为输入与经过验证的输出之间的契约。为相关成果命名，定义成功检测标准，并拒绝默许的半完成状态。

对于会消耗资金或修改生产数据的操作，必须经过人工审批。编译时的配置并不等同于业务功能的完整性。

编写一份简短的操作手册：说明如何轮换密钥、如何清空队列、以及如何回滚上一次的数据导入操作。

在记录功能测试结果的同时，也要标注相应的处理时间以及令牌或查询的成本。提前了解成本情况，可以避免在系统从演示环境过渡到共享环境时出现意外账单。

对于会消耗资金或修改生产数据的操作，必须经过人工审批。编译时的配置并不等同于业务功能的完整性。

在升级整个技术栈之前，应先冻结现有版本，为关键流程保存完整的操作记录，并确认好回滚步骤。共享环境需要设置速率限制、进行租户身份验证，同时还要明确负责密钥轮换的人员。与其追求华而不实的临时演示，不如注重扎实可靠的稳定性。

关于11b06b04feeb的批量处理说明：不要将提供者密钥放入代码仓库，为每个会话设置令牌使用上限，并将转录内容存储在评估测试用例的旁边，以便后续更换模型时仍能保持可比性。

在处理强化安全性的第0阶段时，首先需明确相关规范：所需输入、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改始终符合要求。同时需将正常流程与故障恢复流程一并记录下来。重试机制、人工审核环节以及错误处理方式都是产品本身的组成部分，而非后续需要补充的内容。

强化安全性细节0/898：需统计该处理步骤的运行时间、错误类型以及令牌消耗情况，然后依据固定的评估标准而非个人经验来决定是否保留该变更。

在将加固过程视为可测量的表面时，第一阶段的效果最佳。在扩大范围之前，需记录一份理想的测试结果、一个故障案例以及回滚说明。 应将此阶段视为输入与经过验证的输出之间的契约。为相关成果命名，明确成功标准，绝不允许默许不完整的处理方式。

加固细节 1/898：需测量此阶段的执行时间、错误类型以及令牌消耗情况，然后依据固定的评估标准而非主观判断来决定是否保留该变更。

在强化措施的第2阶段，应在修改代码之前明确输入参数、该步骤的负责人以及完成标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 配置信息应置于应用程序代码之外。环境文件、密钥存储以及功能标志应集中存放于一个位置，这样操作人员无需查看整个系统结构即可进行审计。

强化措施细节2/898：需统计该措施的墙钟时间、错误类型以及令牌消耗情况，然后根据固定的评估标准而非主观判断来决定是否保留该变更。

在处理强化措施的第3阶段时，首先写下相关约定：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改始终符合既定标准。 相比冗长的脚本，更应采用小型且可测试的单元。当某个步骤失败时，故障应能指向单一责任点，而非复杂的流程链。

强化措施细节3/898：需测量该步骤的运行时间、错误类型以及代币消耗情况，然后依据固定的评估标准而非主观判断来决定是否保留该修改。

将强化措施的第4阶段视为可度量的对象来处理效果最佳。在扩大范围之前，先记录一份理想的操作流程、一个失败案例以及回滚说明。 在功能结果旁同时记录时间消耗及代币或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外支出。

强化措施细节4/898：为该记录测量运行时间、错误类型以及令牌消耗情况，然后依据固定的问题集而非个人经验来判断是否保留该变更。

对于强化措施的第5阶段，在修改代码之前需明确输入参数、该步骤的负责人以及完成标准。操作人员应能够从已知的检查点重新执行该步骤，而无需猜测隐藏状态。同时需将正常流程与故障恢复流程一并记录下来。重试机制、人工审核环节以及错误处理方式都是产品本身的组成部分，而非后续需要补充的内容。

强化措施细节5/898：为该记录测量运行时间、错误类型以及令牌消耗情况，然后依据固定的问题集而非个人经验来判断是否保留该变更。

在将加固措施视为可测量的表面时，第0阶段的效果最佳。在扩大范围之前，先记录一份理想的运行结果、一个故障案例以及回滚说明。 在功能测试结果旁同时记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在系统从演示环境过渡到共享环境时出现意外费用。

加固细节0/917：为该步骤测量实际执行时间、错误类型以及令牌消耗情况，然后依据固定的评估标准而非主观经验来决定是否保留该变更。

对于加固措施的第1阶段，在修改代码之前需明确输入参数、该步骤的负责人以及完成标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测系统的隐藏状态。 需同时记录正常流程和故障恢复流程。重试机制、人工审核环节以及错误处理方式都是产品本身的组成部分，而非后续需要补充的内容。

强化细节 1/917：测量该笔记的运行时间、错误类型以及令牌消耗情况，然后依据固定的问题集而非个人经验来决定是否保留该变更。