跳到主要内容
Super Durable
菜单

在生产环境运行 Agent

查看 Markdown

Agent Session 和其他所有 Flow 运行在同一个 Dex Server 上。把 Session 交给用户使用之前，逐项检查下面的内容。每一行都链接到包含详情的页面。

事项需要知道什么详情
Dex ServerDex Server 运行在 Dex Cloud 上，或者运行在你自己带对象存储的基础设施上。一个集群由一个活动节点提供服务。选择 Dex 的运行方式、在生产环境运行 Dex
认证Public API 认证默认关闭。在 server 能被其他机器访问之前打开它。Authentication（英文）
保留期已关闭的 Session 默认在 7 天内仍可读取，之后 server 可能删除它。需要保存更久的对话记录，请放在你自己的存储中。Retention（英文）
Session 大小Session 的 active state 达到 state 上限（默认 64 MiB）时，Session 结束。Message 保留策略，以及 harness 支持时的 compaction，会让它保持在上限以下。Session limits（英文）
模型花费在每次模型调用前后，按账号或 API key 检查花费；Dex 把花费上限留给你的代码。Spend limits（英文）
Credential把模型 API key 和 tool credential 放在 Worker 的环境或 secret store 中，永远不要放进 Session input 或 message。Credentials（英文）
审批期限调用不能无限期等待时，设置一个期限。没有期限的审批会一直等到有人答复，turn 也随之等待。Approval deadlines（英文）
升级部署在下一次模型调用或 Tool Call 时生效。有些版本不能与上一个版本同时运行：启动新 Worker 之前，先停掉所有旧 Worker。这样的停止会让正在运行的单次尝试 tool 变成 outcome_unknown，并让 RPC 以 Unavailable 失败，直到 Worker 恢复；请先优雅地停止 Worker。Upgrades（英文）
Worker每个正在运行的模型调用、Tool Call 和 RPC 都占用一个 Worker task slot。按并行 Tool Call 的数量确定 Worker 规格。Workers and timeouts（英文）
流式传输AgentEvents Stream 把实时文本送到你的 UI。按你的流量确定它的规格。AgentEvents Stream（英文）
TracingSession 可以为每次模型调用和 Tool Call 输出 OpenTelemetry span。内容采集默认关闭。Tracing（英文）
部署形态Web replica 持有 Client。Worker 进程运行 Session Flow。部署形态
部署期间的调用send_user_message 和 resolve_interrupt 是 RPC。它们需要正在运行的 Worker，而重试的 send_user_message 可能把同一条消息排队两次。部署期间的 RPC
未知结果被崩溃打断的写操作会记录为 outcome_unknown，并且不会再次运行。决定由谁来核对它。核对 outcome_unknown

部署形态

生产环境有三类进程：

  • Web app replica 持有 Client。它们启动 Session、发送消息并流式返回回复。它们需要 Session Flow 的定义来编码调用，但从不运行 Step。
  • Worker 进程 运行 Session Flow：每次模型调用、Tool Call 和审批等待。对小应用来说，web 进程也可以同时托管 Worker，Agent 快速开始程序就是这样做的。
  • Dex Server 保存每个 Session。Worker 主动向外连接它，所以不需要入站端口。

滚动部署期间，每个 Step 运行的是接手它的那个 Worker 的代码。正在进行的 turn 会用新代码完成它的下一次模型调用或 Tool Call。缺少某个 Step 类型的 Worker 会拒绝接手，由另一个 Worker 接手。少数版本不能与上一个版本同时运行，需要先停掉所有旧 Worker；Upgrades release notes（英文）列出了这些版本。

部署期间的 RPC

应用通过 RPC 与 Session 通信，例如 send_user_message 和 resolve_interrupt。RPC 在 Worker 上运行。它从不在队列中等待：如果 10 秒内没有 Worker 接手，调用就会以 Unavailable 失败。在停掉所有 Worker 的部署期间，这些调用会一直失败，直到新的 Worker 连上。Session 本身不会丢失任何东西。

  • 遇到 Unavailable 时重试。 用 backoff 重试，直到 Worker 恢复。以 Unavailable 失败的调用什么都没有改变。
  • resolve_interrupt 可以安全重试。 已经生效的审批第二次会返回 False，什么都不会再次运行。
  • send_user_message 可能发送两次。 Dex 会再次运行重复的 RPC。如果响应丢失，重试可能把同一条消息排队两次。重新发送之前，用 get_agent_snapshot 读取 Session，在 messages 和 queued_messages 中查找这条消息。如果它排队了两次，用 delete_queued_message 删除多出的副本。

下面这个辅助函数就是这样做的。每个标签页都展示完整模块。它重试可能没有送达的发送，每次重试前检查 Session，并删除多出的排队副本。它按文本区分消息，所以用户每发一条消息就调用它一次。

safe_send.pyPython
"""Sends a user message to a session once, even when a send must be retried."""
from __future__ import annotations

import asyncio
from typing import Protocol

import grpc  # type: ignore[import-untyped]
from dex import DexServiceError
from dex_agents import AgentSnapshot, MessageRole

# A lost response leaves a send's outcome unknown; these codes mean "check first".
UNCERTAIN_CODES = (grpc.StatusCode.UNAVAILABLE, grpc.StatusCode.DEADLINE_EXCEEDED)

class MessageSessions(Protocol):
    async def get_agent_snapshot(self, session_id: str) -> AgentSnapshot: ...

    async def send_user_message(self, session_id: str, text: str) -> bool: ...

    async def delete_queued_message(self, session_id: str, message_id: str) -> bool: ...

async def send_user_message_once(
    sessions: MessageSessions,
    session_id: str,
    text: str,
    *,
    attempts: int = 5,
    retry_delay_seconds: float = 1.0,
) -> None:
    """Queues one user message, retrying a failed send without queuing it twice.

例子: examples/python/agent-harnesses/agent_harness_examples/quick_start/safe_send.py

其他可能重复的请求，参见 限制。

下一步

  • 限制：Dex 整体的默认值和已知不足。
  • Session：每个 Step 给一个 turn 增加了什么。