如何让 AI Agent 的生产崩溃自动恢复？Duralang 用一个 @dura 装饰器解决韧性难题

AI Agent 在生产环境的崩溃，大部分时候不是模型不够聪明——是基础设施不给力。

模型选对了工具、推理路径完全正确，但一个网络超时就能让整个 pipeline 功亏一篑：所有已完成步骤丢失，Agent 需要从头开始。想象一个多步骤调研 Agent，在跑完 15 个 API 调用后因为第 16 个超时就要重来——这种挫败感在做 AI Agent 开发的朋友一定不陌生。

问题是，现有的持久化方案都是为确定性程序设计的——它们假设执行路径是固定的。但 LLM 驱动的 Agent 是随机的：每一步调用哪个工具、调用几次、按什么顺序，都由模型在运行时决定。传统方案（如 Temporal 原生 SDK）需要你手写 @workflow.defn、@activity.defn、手动序列化状态——对于一个由随机执行图驱动的 AI Agent 来说，这种侵入性改造成本高得几乎不现实。

Duralang 就是来解决这个矛盾的。它用 Python 装饰器的方式，把 LangChain Agent 的每一次调用都变成 Temporal Activity，而你只需要加一行 @dura。

一行代码，生产级韧性

from duralang import dura, dura_agent

@dura
async def my_agent(messages):
    agent = dura_agent(
        model="claude-sonnet-4-6",
        tools=[web_search, calculator],
    )
    result = await agent.ainvoke({"messages": messages})
    return result["messages"]

去掉 @dura → 普通的 LangChain Agent。加上 @dura → 具备完整持久化能力的随机执行图。代码量差异：一行。

自动获得的功能栈

加了 @dura 装饰器后，你的 Agent 自动获得以下能力，零配置：

每个 LLM 调用 → Temporal Activity（自动重试超时/限流）

每个工具调用 → Temporal Activity（独立心跳，独立重试）

每个 MCP 调用 → Temporal Activity（通过 langchain-mcp-adapters）

Agent 调用 Agent → Temporal Child Workflow（每层独立持久化）

实际运行效果

当你在本地运行 Duralang Agent 时，Temporal UI（默认 localhost:8233）会实时显示完整的执行时间线——每次 LLM 调用、工具调用、MCP 调用的输入、输出、耗时和重试次数，全部一目了然。

orchestrator              ← 随机、不可摧毁
├── dura__llm             ✅ 持久化
├── researcher            ← Child Workflow，独立事件历史
│    ├── dura__llm        ✅
│    └── dura__tool       ✅
└── analyst               ← Child Workflow，独立事件历史
     ├── dura__llm        ✅
     └── dura__tool       ✅

每一层都是随机的——LLM 独立决定下一步。每一层也是独立持久的——如果 analyst 崩溃了，只有 analyst 需要重试，researcher 的已完成工作毫发无损，orchestrator 的进度全部保留。

自定义配置

默认值覆盖大部分场景，但也可以通过 DuraConfig 精细调优：

from duralang import dura, DuraConfig, ActivityConfig
from datetime import timedelta
from temporalio.common import RetryPolicy

config = DuraConfig(
    task_queue="agents-prod",
    llm_config=ActivityConfig(
        start_to_close_timeout=timedelta(minutes=3),
        heartbeat_timeout=timedelta(seconds=30),
        retry_policy=RetryPolicy(maximum_attempts=5),
    ),
    tool_config=ActivityConfig(
        start_to_close_timeout=timedelta(minutes=1),
        retry_policy=RetryPolicy(maximum_attempts=4),
    ),
)

@dura(config=config)
async def my_agent(messages):
    ...

免费的可观测性

Duralang 的一个隐藏好处是：你获得了一个免费、自托管的 AI Agent 可观测平台。

Temporal UI 展示每个 Agent 的完整执行历史——包括 LLM 调用的输入输出、工具调用的结果、执行耗时和重试次数。不需要 LangSmith 订阅，不需要第三方服务，数据不会离开你的基础设施：

temporal server start-dev

python my_agent.py

安装与快速上手

pip install duralang

一条命令，零依赖冲突。Duralang 自动处理 Temporal Worker 的注册、Activity 的包装和事件历史的记录——你完全不需要了解 Temporal 的 SDK 细节。

更深入的用法参考 GitHub 仓库（附 YouTube 演示视频），或者直接看 Temporal 官方的 Code Exchange 页面。

总结与推荐

Duralang 解决了一个真实且被普遍忽略的问题：AI Agent 的运行时可靠性。它用一种极轻量的方式（一行 @dura）把 Temporal 的工业级持久化能力注入了 LangChain 生态，让开发者可以在不修改 Agent 逻辑的前提下获得生产级的崩溃恢复能力。

如果你正在构建面向用户的 AI Agent 服务——尤其是涉及多步骤推理、多个工具调用、或长时间运行的 Agent——Duralang 是一个值得加入技术栈的工具。