--- title: "在生产环境运行 Agent" description: "在生产环境运行 Agent Session 之前的检查清单：你运行的 server、认证、保留期、Session 与花费上限、升级、tracing、审批期限和 Worker 规格。" canonical_url: "https://docs.superdurable.io/zh-Hans/agent-frameworks/production/" source: "docs/i18n/zh-Hans/docusaurus-plugin-content-docs/current/agent-frameworks/production.mdx" locale: "zh-Hans" language: "python" --- > 供 AI Agent 使用的全部页面索引：https://docs.superdurable.io/zh-Hans/llms.txt # 在生产环境运行 Agent 本页代码为 Python。其他语言：[TypeScript](https://docs.superdurable.io/zh-Hans/agent-frameworks/production.typescript.md) Agent Session 和其他所有 Flow 运行在同一个 Dex Server 上。把 Session 交给用户使用之前，逐项检查下面的内容。每一行都链接到包含详情的页面。 | 事项 | 需要知道什么 | 详情 | | ---------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------- | | Dex Server | Dex Server 运行在 Dex Cloud 上，或者运行在你自己带对象存储的基础设施上。一个集群由一个活动节点提供服务。 | [选择 Dex 的运行方式](https://docs.superdurable.io/zh-Hans/components/choose-how-to-run.md)、[在生产环境运行 Dex](https://docs.superdurable.io/zh-Hans/components/production.md) | | 认证 | Public API 认证默认关闭。在 server 能被其他机器访问之前打开它。 | [Authentication](https://docs.superdurable.io/zh-Hans/reference/server/authentication.md)（英文） | | 保留期 | 已关闭的 Session 默认在 7 天内仍可读取，之后 server 可能删除它。需要保存更久的对话记录，请放在你自己的存储中。 | [Retention](https://docs.superdurable.io/zh-Hans/reference/server/configuration.md#retention)（英文） | | Session 大小 | Session 的 active state 达到 state 上限（默认 64 MiB）时，Session 结束。Message 保留策略，以及 harness 支持时的 compaction，会让它保持在上限以下。 | [Session limits](https://docs.superdurable.io/zh-Hans/reference/agents/session-limits.md)（英文） | | 模型花费 | 在每次模型调用前后，按账号或 API key 检查花费；Dex 把花费上限留给你的代码。 | [Spend limits](https://docs.superdurable.io/zh-Hans/reference/agents/spend-limits.md)（英文） | | Credential | 把模型 API key 和 tool credential 放在 Worker 的环境或 secret store 中，永远不要放进 Session input 或 message。 | [Credentials](https://docs.superdurable.io/zh-Hans/reference/agents/credentials.md)（英文） | | 审批期限 | 调用不能无限期等待时，设置一个期限。没有期限的审批会一直等到有人答复，turn 也随之等待。 | [Approval deadlines](https://docs.superdurable.io/zh-Hans/reference/agents/approvals.md#approval-deadlines)（英文） | | 升级 | 部署在下一次模型调用或 Tool Call 时生效。有些版本不能与上一个版本同时运行：启动新 Worker 之前，先停掉所有旧 Worker。这样的停止会让正在运行的单次尝试 tool 变成 `outcome_unknown`，并让 RPC 以 **Unavailable** 失败，直到 Worker 恢复；请先优雅地停止 Worker。 | [Upgrades](https://docs.superdurable.io/zh-Hans/reference/agents/upgrades.md#in-flight-calls)（英文） | | Worker | 每个正在运行的模型调用、Tool Call 和 RPC 都占用一个 Worker task slot。按并行 Tool Call 的数量确定 Worker 规格。 | [Workers and timeouts](https://docs.superdurable.io/zh-Hans/reference/agents/workers.md)（英文） | | 流式传输 | AgentEvents Stream 把实时文本送到你的 UI。按你的流量确定它的规格。 | [AgentEvents Stream](https://docs.superdurable.io/zh-Hans/reference/agents/event-stream.md)（英文） | | Tracing | Session 可以为每次模型调用和 Tool Call 输出 OpenTelemetry span。内容采集默认关闭。 | [Tracing](https://docs.superdurable.io/zh-Hans/reference/agents/tracing.md)（英文） | | 部署形态 | Web replica 持有 Client。Worker 进程运行 Session Flow。 | [部署形态](https://docs.superdurable.io/zh-Hans/agent-frameworks/production.md#deployment-shape) | | 部署期间的调用 | `send_user_message` 和 `resolve_interrupt` 是 RPC。它们需要正在运行的 Worker，而重试的 `send_user_message` 可能把同一条消息排队两次。 | [部署期间的 RPC](https://docs.superdurable.io/zh-Hans/agent-frameworks/production.md#rpcs-during-a-deploy) | | 未知结果 | 被崩溃打断的写操作会记录为 `outcome_unknown`，并且不会再次运行。决定由谁来核对它。 | [核对 outcome_unknown](https://docs.superdurable.io/zh-Hans/agent-frameworks/tool-calls.md#reconcile-outcome-unknown) | ## 部署形态 生产环境有三类进程： > **图示**：用户访问 Web 应用副本，它们持有 Client，向 Dex Server 发送 RPC 和读取请求。Worker 进程通过出站连接从 Dex Server 拉取任务，运行 Session Flow：模型调用、Tool Call 和审批。Dex Server 把 Session 状态保存在对象存储中。 - **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](https://docs.superdurable.io/zh-Hans/reference/agents/upgrades.md#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，并删除多出的排队副本。它按文本区分消息，所以用户每发一条消息就调用它一次。 **Python** ```python """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. Before each retry it reads the session, and stops when a new copy of the text is there. At the end it deletes any extra queued copy. It tells messages apart by their text, so call it once for each message a user sends. Raises: DexServiceError: When no attempt reached a Worker, or on any other error. """ copies_before = _copies(await sessions.get_agent_snapshot(session_id), text) for attempt in range(1, attempts + 1): try: await sessions.send_user_message(session_id, text) break except DexServiceError as error: if error.code not in UNCERTAIN_CODES or attempt == attempts: raise await asyncio.sleep(retry_delay_seconds) if _copies(await sessions.get_agent_snapshot(session_id), text) > copies_before: break await _delete_extra_copies(sessions, session_id, text, copies_before + 1) async def _delete_extra_copies( sessions: MessageSessions, session_id: str, text: str, wanted: int ) -> None: snapshot = await sessions.get_agent_snapshot(session_id) extra = _copies(snapshot, text) - wanted queued = [item for item in snapshot.queued_messages if item.message.text == text] if extra <= 0: return for item in reversed(queued[max(0, len(queued) - extra) :]): await sessions.delete_queued_message(session_id, item.message_id) def _copies(snapshot: AgentSnapshot, text: str) -> int: delivered = sum( 1 for message in snapshot.messages if message.role == MessageRole.USER and message.text == text ) queued = sum(1 for item in snapshot.queued_messages if item.message.text == text) return delivered + queued ``` 例子: [examples/python/agent-harnesses/agent_harness_examples/quick_start/safe_send.py](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses/agent_harness_examples/quick_start/safe_send.py) 其他可能重复的请求，参见 [限制](https://docs.superdurable.io/zh-Hans/intro/limits.md#requests-that-can-repeat)。 ## 下一步 - [限制](https://docs.superdurable.io/zh-Hans/intro/limits.md)：Dex 整体的默认值和已知不足。 - [Session](https://docs.superdurable.io/zh-Hans/agent-frameworks/sessions.md#latency-and-cost)：每个 Step 给一个 turn 增加了什么。