OpenSRE Gateway 架构解析AI SRE 代理的多通道消息网关设计与运行机制【免费下载链接】opensreBuild your own AI SRE agents. The open source toolkit for the AI era.项目地址: https://gitcode.com/GitHub_Trending/op/opensreOpenSRE 是一个开源的 AI SRE 代理工具包。本文聚焦其Gateway消息网关子系统一个常驻进程如何把 Slack、Telegram、Discord、Buzz 等多个聊天平台统一接入同一个 AI 代理执行引擎以及它背后单进程、单 turn 引擎、多通道的设计约束。读完本文你将掌握网关的入口、生命周期、依赖边界、容量治理与多租户会话绑定机制并能基于源码与测试理解其 fail-closed 与唯一执行引擎的架构决策。网关在 OpenSRE 中的角色Gateway 是 OpenSRE 面向即时通讯平台的统一入口它把 Slack、Telegram、Discord、Buzz 的入站消息接收下来经过鉴权、会话解析与 turn 分发送入同一个无头代理Headless Agent执行引擎再通过各平台的 turn 输出把结果送回用户。用包内开发指南gateway/AGENTS.md的话说这个包承载了进程daemon、传输通道transports、Web 应用与存储四类东西而整个设计反复强调一个核心原则全仓库只有一个 turn 引擎TurnRunner所有聊天通道都是它的入站适配器任何通道不得自建第二条执行回路。入口点先看这几个文件gateway/AGENTS.md给出了先打开这些文件的入口点清单结合源码整理如下角色路径生产入口slash 端口注入CLI 命令opensre gateway start/opensre gateway start --foreground组合根在包外包 maingateway/main.py ——fail closed无 slash 端口胶水不是生产入口进程组合根gateway/core/lifecycle/controller.pyGatewayController注入slash_ports_factory表面启动Web chat 组合器gateway/startup.pystart_gateway/StartedGateway守护进程pidfile spawngateway/core/process/supervision.py —— 调用方传入 argv从不指名 CLI 或surfaces.gateway_entryTurn 回调infrastructure/turn_host/turn_runner.pyTurn 契约输出、回调infrastructure/turn_host/turn_output.py、infrastructure/turn_host/turn_callback.py传输注册表名称、注册、workergateway/transports/names.py、gateway/transports/registration.py、gateway/transports/startup.pyTurn 中间件决策、策略、审批、停止、锁gateway/core/middleware/配置 / 传输错误gateway/core/lifecycle/errors.pyGatewayConfigurationError、GatewayTransportFailedErrorWeb 表面FastAPIgateway/web/webapp.pyappTelegram / Slack / Discord / Buzz 启动gateway/transports/telegram/startup.py、gateway/transports/slack/startup.py、gateway/transports/discord/startup.py、gateway/transports/buzz/startup.py为什么python -m gateway会直接报错退出gateway/main.py 的 docstring 写得非常明确这个模块故意不是生产组合根。slash 命令端口SlashPortsFactory由包外的 CLI 组合根注入直接python -m gateway启动会让 chat 在没有slash_invoke的情况下裸奔。因此它直接抛出python -m gateway has no slash-port glue and is not a production entry. Start the gateway with: opensre gateway start or: opensre gateway start --foreground同样gateway/core/lifecycle/controller.py 顶层的start_gateway()兼容包装函数也要求必须传入slash_ports_factory否则抛SystemExit(_BARE_MANAGER_EXIT)。这是典型的fail-closed设计宁可拒绝启动也不允许缺依赖的半成品进程上线。单元测试则可以绕过该约束直接构造GatewayController(...)。生命周期GatewayController组合根网关进程的启动顺序在GatewayController.start_gateway中一目了然见 gateway/core/lifecycle/controller.pystart_gateway() → configure_process(GATEWAY_PROFILE) # bootstrap.process环境配置 → compose turn runner # 唯一的 TurnRunner 容量门 → start_surfaces() # 委托 gateway/startup.pyweb chat → start_scheduler() # 承载 infrastructure.scheduling.scheduler → ready # 写 pidfile、组件状态、log [gateway] ready关键细节凭证先行_load_credentials在任何传输、调度器或 worker 启动前完成凭证水合credential hydration失败则抛GatewayConfigurationError。一个 TurnRunner控制器只构造一个TurnRunner携带gateself.turn_gate容量门与admission_checkadmit_metered_turn计费准入所有聊天传输共用它注释明确警告不要在它外面再包一层 turn runner。表面状态start_surfaces委托 gateway/startup.py 的start_gateway()一并启动 Web 服务器与全部聊天传输返回StartedGateway持有 web 句柄、传输句柄与各组件状态。缺少聊天凭证的传输记为not configured就绪或运行失败记为failed其余照常启动——一个平台的故障不会拖垮整个网关。调度器是平台组件start_scheduler()承载的是infrastructure.scheduling.scheduler通过scheduler_runners().gated(self.turn_gate)安装受容量门约束的 runner 并启动后台调度器。它不是消费表面也不是传输因此没有gateway/scheduler/包也不得把 runner 代码挪进gateway/core/lifecycle/。可用OPENSRE_GATEWAY_HOST_SCHEDULERfalse关闭进程内调度器改由独立的MODEscheduler服务承载避免两个进程重复触发定时任务。守护进程pidfile spawngateway/core/process/supervision.py 提供的是进程监督而非调度start_gateway_daemon(argv...)把调用方传入的 argv 作为子进程 spawn输出落~/.opensre/gateway/gateway.logPID 记入gateway.pid启动窗口内子进程死亡则返回失败并附日志尾部。子进程 argv 由表面持有源码形态下是python -m surfaces.gateway_entry冻结形态下是opensre gateway start --foreground该模块从不自己指名组合根。包布局与依赖规则网关包按core/agent_harness/prompts/的惯例拆分核心基础设施 vs 对等表面 vs 组合器。core/—— 进程与叶子基础设施process、lifecycle、storage、billing、attachments、session、config。不 import transports 或web只有core/lifecycle/controller.py允许 importgateway.startup。startup.py—— 门面把 web chat 作为一个消费集合通过transports/startup.py它持有注册表并只 import 各对等方的startup统一启动。transports/—— 聊天对等方slack、discord、telegram、buzz。每个对等方自持设置、入站 worker、安全、turn 输出与startup.py。对等方之间永不互 import也永不 importgateway.startup/web两者都需要的公共能力放在core/逐 turn 步骤在gateway.core.middleware或infrastructure.turn_hostturn runner、turn 输出、会话代理池。web/—— Web 表面FastAPI 健康检查应用与告警摄取。可 importcore/不得 import 聊天传输或gateway.startup。无环依赖规则core.controller → startup.py → transports/startup.py → transports.{telegram,slack,discord,buzz}.startup → web peer transports · web → core leaves (对等方永不互 import、不 import gateway.startup、不 import 彼此包)这套 DAG 与对等隔离由边界测试钉死包 DAG 由 gateway/tests/test_package_borders.py 保障表面允许面则由 tests/shared/test_surface_border.py 以精确 allowlist钉死——目前仅core.process.supervision、core.lifecycle.controller、web.web_server三个模块对表面可见扩大它属于刻意的架构变更而非新增 import。另外infrastructure.scheduling.scheduler不得 importTurnRunner由tests/test_package_borders.py::test_scheduler_never_imports_the_gateway_turn_runner钉死。存储层约定core/storage中open_database()为进程提供唯一一份已迁移的PostgresDatabase无DATABASE_URL时为None每个领域的repository.py带一个选择器返回 Postgres 或进程本地实现。宿主只调用选择器不自行构造 store。core/storage/session/resolver.py的SessionResolver负责按平台键platform chat_id做会话绑定把 create / resolve / rotate 委托给SessionManager。Channel vs Producer两条进入代理的路径gateway/AGENTS.md用一个表格界定了工作如何到达代理的两种方式并警告混用二者正是出现第二个 turn 引擎的根源。| | 是否有用户与 turn 输出 | 入口 | |--|------------------------|--------| |ChannelSlack、Telegram、Discord、Buzz | 是 |TurnRunner——(text, session, output, logger)| |Interactive shell| 是 |现状HeadlessAgent.handleAgentBuildConfig。目标与其他 channel 一样走chat动词——它有用户和 turn 输出规则本就覆盖它。构建配置已共享turn 入口尚未统一。 | |Producerinfrastructure.scheduling.scheduler、定时 digest/PR runner | 否 | 内嵌AgentSession.run_headless_turn|代理构建钩子位于core.agent_harness.agent_build_config.AgentBuildConfig不在传输注册表也不是宿主的再导出。chat 省略构建配置由会话代理池注入网关能力扣减ensure_gateway_capability_policyshell 则设置自己需要的构建钩子不设置apply_capability_policy。网关进程可以承载调度器走同一个process_turn_gate但这并不使调度器成为 channel——调度器必须走run_headless_turn不得触碰TurnRunner。唯一 facade 动词chat网关对外只暴露一个 turn 动词动词入口形态获得的能力chatTurnRunner.__call__(text, session, output, logger)返回None所有结果通过 turn 输出送达用户容量门、能力策略、SessionAgentPool复用、审批、取消控制台、身份策略、turn 超时、终态结果、at-capacity 文案任何拥有活跃用户和 turn 输出的东西都用chat——这正是 interactive shell 正在迁移到的规则。为什么 chat 返回None四个聊天传输是fire-and-forget的turn 输出本身就是回复路径没有需要回传的值。而像 shell 这样需要把 turn 结果作为值使用的调用方想要TurnResult做记账、给 prompt recorder、读final_intent这个签名就服务不了。指南明确要求拓宽签名是对全部四个传输的契约变更应当刻意决策而不是在TurnRunner旁边再加第二个入口。一次消息的完整流转每个聊天传输共用一个TurnRunner容量门可选。Slack、Discord、Telegram 的 dispatcher 只是入站适配器授权 → 解析会话 → 构造 turn 输出 → 调用共享回调禁止再添加第二个生产 turn-runner 类。具体执行在 infrastructure/turn_host/turn_runner.pyTurnRunner.run先取turn_slot(self._gate)——容量门拒绝时直接output.finalize(AT_CAPACITY_MESSAGE)返回NoneOpenSRE is at capacity. Please try again shortly.。准入检查admission_check如计费admit_metered_turn在槽位内执行计费钩子不应为容量本会拒绝的工作收费。进入_run_turn建立取消事件与CancelConsole在bound_usage_context与traced_session中执行。通过self._pool.session_agent(session..., output..., logger...)从SessionAgentPool取该会话的代理持会话级锁直至 turn 结束然后调用agent.handle(text, TurnBinding(...))——与 interactive shell 完全相同的调用见 infrastructure/turn_host/session_agents.py。结果通过output.finalize(outbound_text or EMPTY_RESPONSE_MESSAGE)写回通道SessionManager.for_session(session).flush(session)持久化session_goal按 surface 上报gateway_turn_started/completed/failed分析事件。回调契约为四参数infrastructure/turn_host/turn_callback.py 定义了契约TurnCallback Callable[[str, SessionCore, TurnOutput, logging.Logger], None]。指南强调不要把chat_id放进这个契约——transport 细节由 output 拥有。动作工具的解析也必须每 turn 从该聊天实时会话经DefaultToolProvider(session, console)进行与 interactive shell 一致不得在进程启动时预计算工具。日志在进程启动时只配置一次GatewayController.start_gateway中的configure_logging不存在长驻的网关级Agent每条入站消息经SessionResolver拿到 per-chatSession后走 headless dispatchcore.agent_harness.turns.headless_agent。多租户principal / actor 与存储作用域Slack、Discord、Telegram 各自在transports/peer/principal.py中解析自己的StorageScope对等隔离——无跨 import。Principal是组织 siloORGANIZATION_ID缺失则 fail closed。Actor是平台用户 id。turn 在bound_storage_scope下运行绑定、会话与集成落到组织挂载点或~/.opensre/orgs/id/。CLI / interactive shell 保持未绑定遗留的空 principal/actor id。SessionResolver会把同文档的遗留空 id 行一次性领养进作用域键adopt_unscoped_binding会移除未作用域行从而第二个 actor 无法继承那个会话见 gateway/core/storage/session/resolver.py 的_lookup_or_adopt_session_id。注意resolve()的principal/actor是有意做成必填参数的它们曾默认为None导致传输忘传时作用域绑定静默退化为遗留空键行真正想用遗留键的调用方必须显式传None。容量治理进程门 vs 传输池指南先划清边界云级横向扩展通过新增 Fargate task 实现无限容量是这两层之上的第三层——每个 task 是一个带自己门的网关进程饱和时提升舰队规模/worker不要放开进程内门的限制。进程内有两个不同的限制切勿混为一谈层机制满载行为进程TurnConcurrencyGate/process_turn_gate()由OPENSRE_SIZE_PROFILE决定SMALL1、MEDIUM2、LARGE4Chat非阻塞try_acquire繁忙即丢。调度器 runner阻塞acquire已认领的工作等待。Per-transportmax_concurrent_turns默认同 profile 限制来自turn_limit_for_profile可用*_GATEWAY_MAX_CONCURRENT覆盖限制该传输在命中共享 turn runner 之前可并行处理的入站消息数。不能替代进程门。Telegram/Slack/Discord ──► TurnRunner.try_acquire ──► process_turn_gate() Scheduler (agent runners) ──► blocking acquire ──► same gate生产 chat 容量在TurnRunner(gatecontroller.turn_gate)上GatewayController使用infrastructure.turn_host.concurrency.process_turn_gate。ConcurrencyLimitedTurnHandler仅用于测试生产只允许TurnRunner上的gate见 infrastructure/turn_host/concurrency.py。OPENSRE_MAX_CONCURRENT_TURNS允许在不升级 size profile 的情况下提高并发turn 是 I/O 密集的非正数或不可解析的值会被忽略并告警避免拼写错误把门降下来。chat 分析事件使用gateway_turn_*且surface ∈ {slack, telegram, discord}。Agent 生命周期每逻辑会话一个代理每个逻辑聊天会话只构造一个HeadlessAgent由SessionAgentPool持有然后服务多次 turn。每条入站消息BindableOutput.bind(outer_gateway_output)—— 把会话输出绑到代理上每次 turn 的传输目的地都可能变化。bind_turn(session…, accounting…, console…, tool_hooks…)—— 会话 / 取消 / 审批。不要在这里传output除非是要替换OutputSink对象本身那样OutputBindable端口如 reasoning也必须跟随。agent.handle(text, TurnBinding(...))—— SessionGoal turn 循环 每 turndispatch。不要在网关路径上把它包成AgentSession.chat池拥有代理并直接调用handle。不要每条消息都新建无头代理。同会话 turn 在池的 per-session 锁上串行不同会话在容量门下保持并发见SessionAgentPool.session_agent的上下文管理器锁横跨整个分发过程——重新绑定正是让代理每 turn 特异性的机制提前释放会让下一 turn 重定向一个仍在流式输出的代理。多 turn 定时循环应为一个循环保留一个代理真正的一次性 digest 才用AgentSession.run_headless_turn。通道能力对等host parity四个聊天表面共用同一个 turn 引擎ingress →TurnRunner→SessionAgentPool→agent.handle。能力对等表yes / partial / no / n/a关注点SlackTelegramDiscord取消 / turn 中停止yes—— 软超时 用户/stop经ActiveTurnRegistry→output.turn_cancelyes—— 相同yes—— 相同审批 /before_tool_callyes—— Block Kit approval_tool_hooksyes—— 内联键盘 approval_tool_hooksyes—— components approval_tool_hooks工具解析yes—— 实时DefaultToolProvider(session)yes—— 相同yes—— 相同输出脱敏yes——user_facing_error_messageyes—— 相同yes—— 相同Principal / actoryes——slack/principal.pyyes——telegram/principal.pyyes——discord/principal.py容量门yes—— 进程门 传输池yes—— 相同 TG 信号量yes—— 相同 executor文档化的例外不要修成第二条循环网关 chat 禁用task_cancel/llm_provider两项能力infrastructure.turn_host.capability_policy.ensure_gateway_capability_policy见 capability_policy.py 的UNSUPPORTED_GATEWAY_CAPABILITIES。软 turn 超时与用户/stop/stop//cancel都设置output.turn_cancel让 ReAct 循环 / 剩余工具协作式停止与 shell 的cancel_requested通过CancelConsoleActiveTurnRegistry对齐。编排器跳过 gather/answer 并报告final_intentcli_agent_cancelledEvent 触发后实时输出停止排空流块。/stop在per-conversation turn 锁之外处理以便打断进行中的 turnexecutor 线程不被杀死进行中的 LLM/provider 调用仍会完成当前请求。Telegram 写工具审批要求非空allowed_user_ids允许名单与 Discord 相同的 fail-closed 姿态。测试策略网关单元测试位于包内gateway/tests/树而非仓库级tests/树。要点边界与容量网关边界 并发门测试test_package_borders.py、test_surface_border.py、gateway/tests/runtime/test_concurrency_gate.py、gateway/tests/runtime/test_turn_runner.py。烟雾测试针对gateway/cli.gateway_*标签的本地烟雾套件。Dogfood仅 dev silo在 dev Slack 上mention验证线程连续性、Digging in…、Want me to:→yes且只允许一个 Socket Mode 消费者。注意笔记本上opensre gateway 烟雾 ≠ dogfood不能互相替代。E2E 回归把规范化轮询的 Telegram 消息灌入handle_polled_inbound_telegram_message(...)让其调用 turn handler不要为了验证命令分发而换入假 LLM 客户端——当测试只需验证 provider 与回调管线时优先使用显式注册的命令如/status。小结OpenSRE Gateway 的设计可以浓缩为几条可验证的工程纪律唯一引擎所有聊天通道共用同一个TurnRunnerchannel 走chat动词producer 走run_headless_turn二者物理隔离。Fail closed裸python -m gateway拒绝启动缺slash_ports_factory拒绝启动缺ORGANIZATION_ID拒绝提供租户能力。分层有界core→startup→transports/web的单向 DAG靠边界测试钉死对等方永不互 import。容量分层进程级TurnConcurrencyGate统一约束 chat 与调度器传输级池只做前置限流云级扩展靠增加进程。会话即租户principal/actor 作用域下的会话绑定与一次性遗留数据领养防止跨 actor 会话串扰。测试即文档从gateway/tests/的测试文件能直接读出上述每一条规则的验证方式。继续深入可以阅读 gateway/README.md、gateway/core/AGENTS.md、gateway/transports/AGENTS.md、gateway/web/AGENTS.md 以及 gateway/tests/ 下的各通道测试它们共同构成对这个消息网关最完整的行为规格。【免费下载链接】opensreBuild your own AI SRE agents. The open source toolkit for the AI era.项目地址: https://gitcode.com/GitHub_Trending/op/opensre创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考