# Synapse：基于Elixir的声明式多智能体编排框架

> Synapse是一个无头、声明式的多智能体编排框架，提供领域无关的信号总线、带Postgres持久化的工作流引擎和可配置的智能体运行时，内置代码审查领域作为参考实现。

- 板块: [Openclaw Llm](https://www.zingnex.cn/forum/board/openclaw-llm)
- 发布时间: 2026-04-04T23:45:11.000Z
- 最近活动: 2026-04-04T23:52:59.436Z
- 热度: 163.9
- 关键词: 多智能体, Elixir, OTP, 工作流引擎, 智能体编排, 声明式配置, 代码审查, LLM集成, 信号总线, Postgres持久化
- 页面链接: https://www.zingnex.cn/forum/thread/synapse-elixir
- Canonical: https://www.zingnex.cn/forum/thread/synapse-elixir
- Markdown 来源: ingested_event

---

## 多智能体系统的编排挑战

随着大语言模型能力的快速发展，基于智能体的应用架构正从单一智能体向多智能体协作系统演进。然而，构建可靠的多智能体系统面临着诸多工程挑战：

- **通信协调**：智能体之间如何高效交换信息
- **状态管理**：复杂工作流的执行状态如何追踪和恢复
- **故障处理**：部分失败时如何保证系统整体稳定性
- **可观测性**：如何监控和审计多智能体的协作过程

传统的解决方案往往需要开发者编写大量的协调代码（如GenServer），这不仅增加了开发复杂度，也使得系统难以维护和扩展。

## Synapse框架概述

Synapse是由nshkr.com团队开发的开源多智能体编排框架，基于Elixir语言和OTP（Open Telecom Platform）构建。它采用声明式的设计理念，让开发者通过配置而非代码来定义智能体行为和工作流。

### 核心设计理念

Synapse的设计体现了几个关键原则：

1. **无头架构（Headless）**：不依赖特定的Web框架（如Phoenix），可作为纯OTP应用运行
2. **声明式配置**：通过配置文件定义智能体，无需编写GenServer样板代码
3. **领域无关**：提供通用的信号总线，支持任意领域的多智能体系统构建
4. **持久化保证**：工作流执行状态持久化到Postgres，支持故障恢复和审计追踪

### 技术栈亮点

Synapse充分利用了Elixir生态系统的优势：

- **并发模型**：基于BEAM虚拟机的Actor模型，天然支持高并发和容错
- **监督树**：OTP的监督机制确保进程故障时自动重启
- **热升级**：支持不停机更新智能体配置
- **遥测集成**：内置Telemetry支持，便于监控和可观测性建设

## 架构组件详解

### 1. 信号总线（Signal Bus）

信号总线是Synapse的核心通信机制，采用发布-订阅模式实现智能体间的松耦合通信。

```elixir
# 发布信号
{:ok, _signal} = Synapse.SignalRouter.publish(
  Synapse.SignalRouter,
  :review_request,
  %{
    review_id: "PR-12345",
    diff: git_diff,
    files_changed: 12,
    labels: ["security"],
    intent: "feature"
  },
  source: "/ci/github"
)

# 订阅信号
{:ok, _sub_id} = Synapse.SignalRouter.subscribe(Synapse.SignalRouter, :review_summary)
```

信号总线支持运行时动态注册新主题，使得系统可以灵活扩展新的信号类型而无需重启。

### 2. 编排器运行时（Orchestrator Runtime）

编排器是Synapse的智能体管理核心。与传统方式不同，Synapse的编排器完全通过声明式配置定义：

```elixir
# priv/orchestrator_agents.exs 示例
%{
  id: :coordinator,
  type: :orchestrator,
  actions: [Synapse.Domains.CodeReview.Actions.ClassifyChange],
  orchestration: %{
    classify_fn: &MyStrategies.classify/1,
    spawn_specialists: [:security_specialist, :performance_specialist],
    aggregation_fn: &MyStrategies.aggregate/2,
    negotiate_fn: &MyStrategies.resolve_conflicts/2
  },
  signals: %{
    subscribes: [:review_request, :review_result],
    emits: [:review_summary],
    roles: %{
      request: :review_request,
      result: :review_result,
      summary: :review_summary
    }
  },
  state_schema: [review_count: [type: :non_neg_integer, default: 0]]
}
```

这种声明式方法的优点在于：

- **零样板代码**：无需编写GenServer
- **动态重载**：修改配置后可触发运行时重载
- **类型安全**：配置通过schema验证，提前发现错误

### 3. 工作流引擎（Workflow Engine）

工作流引擎负责执行声明式的工作流规范，支持复杂的依赖管理、重试策略和错误处理。

```elixir
alias Synapse.Workflow.{Spec, Engine}
alias Synapse.Workflow.Spec.Step

spec =
  Spec.new(
    name: :example_workflow,
    description: "Analyze and generate critique with retries",
    metadata: %{version: 1},
    steps: [
      Step.new(
        id: :fetch_context,
        action: MyApp.Actions.FetchContext,
        params: fn env -> %{id: env.input.review_id} end
      ),
      Step.new(
        id: :analyze,
        action: Synapse.Actions.CriticReview,
        requires: [:fetch_context],
        retry: [max_attempts: 3, backoff: 200],
        on_error: :continue,
        params: fn env -> %{diff: env.input.diff, metadata: env.results.fetch_context} end
      ),
      Step.new(
        id: :generate_critique,
        action: Synapse.Actions.GenerateCritique,
        requires: [:analyze],
        params: fn env ->
          review = env.results.analyze
          %{
            prompt: "Summarize issues",
            messages: [%{role: "user", content: Enum.join(review.issues || [], ", ")}],
            profile: :openai
          }
        end
      )
    ],
    outputs: [
      Spec.output(:review, from: :analyze),
      Spec.output(:critique, from: :generate_critique, path: [:content])
    ]
  )
```

工作流引擎的关键特性包括：

- **步骤依赖**：通过`requires`声明步骤间的依赖关系，引擎自动拓扑排序
- **重试机制**：支持指数退避重试，可配置最大尝试次数
- **错误处理**：支持`on_error: :halt`或`:continue`策略
- **参数函数**：步骤参数可以是函数，动态访问环境变量

### 4. 持久化与审计

Synapse将每次工作流执行持久化到Postgres的`workflow_executions`表，记录包括：

- 工作流名称和版本
- 逐步执行的审计追踪
- 累积的中间结果
- 最终执行状态

这使得开发者可以：

- 查询历史执行记录用于调试
- 构建监控仪表盘
- 从失败点恢复执行
- 满足合规审计要求

## 代码审查领域：参考实现

Synapse内置了一个完整的代码审查领域作为参考实现，展示了如何构建实际的多智能体系统。

### 工作流程

1. **信号触发**：CI系统发布`:review_request`信号
2. **协调器分类**：协调器智能体分析变更意图和标签
3. **专家派生**：根据分类结果派生安全专家、性能专家等
4. **并行审查**：各专家并行分析代码变更
5. **结果聚合**：协调器聚合各专家意见，解决冲突
6. **摘要生成**：发布`:review_summary`信号

### 订阅与查询模式

Synapse支持两种结果消费模式：

**推送模式（实时订阅）**：
```elixir
{:ok, _sub_id} = Synapse.SignalRouter.subscribe(Synapse.SignalRouter, :review_summary)

receive do
  {:signal, %{type: "review.summary", data: summary}} ->
    IO.inspect(summary, label: "Review complete")
end
```

**拉取模式（历史查询）**：
```elixir
Synapse.Workflow.Execution
|> where(review_id: "PR-12345")
|> Synapse.Repo.one!()
```

## LLM集成与多提供商支持

Synapse提供了灵活的LLM网关，支持多种提供商：

### 主要集成方式

1. **Altar.AI集成**（推荐）：通过兼容层委托LLM操作
2. **ReqLLM**（已弃用）：基于Req的多提供商HTTP客户端

### 配置示例

```elixir
# 使用Altar.AI
alias Altar.AI.Integrations.Synapse, as: LLM
{:ok, response} = LLM.chat_completion(%{prompt: "Hello"}, profile: :openai)

# 环境变量配置
export OPENAI_API_KEY=sk-...
export GEMINI_API_KEY=ya29....
```

LLM网关支持配置超时、重试策略和提供商特定选项，使得在不同环境间切换变得简单。

## Jido计划编译器

Synapse还集成了Jido计划编译器，可以将Jido.Plan的DAG（有向无环图）编译为工作流规范：

```elixir
alias Jido.Plan
alias Synapse.PlanRunner
alias Synapse.Workflow.Spec

plan =
  Plan.new(context: %{tenant_id: "acme"})
  |> Plan.add(:fetch, {MyApp.Actions.Fetch, %{value: 2}})
  |> Plan.add(:double, {MyApp.Actions.Double, %{value: 4}}, depends_on: :fetch)

{:ok, exec} =
  PlanRunner.run(plan,
    name: :demo_plan,
    outputs: [Spec.output(:result, from: :double, path: [:value])],
    input: %{},
    context: %{request_id: "req_plan_demo"}
  )
```

这种集成使得开发者可以用更直观的方式定义复杂的工作流依赖关系。

## 自定义领域开发

Synapse的领域无关设计使得开发者可以轻松构建自己的多智能体系统：

### 静态配置注册

```elixir
# config/config.exs
config :synapse, Synapse.Signal.Registry,
  topics: [
    ticket_created: [
      type: "support.ticket.created",
      schema: [
        ticket_id: [type: :string, required: true],
        customer_id: [type: :string, required: true],
        subject: [type: :string, required: true],
        priority: [type: {:in, [:low, :medium, :high]}, default: :medium]
      ]
    ]
  ]
```

### 运行时动态注册

```elixir
Synapse.Signal.register_topic(:my_event,
  type: "my.domain.event",
  schema: [id: [type: :string, required: true]]
)
```

这种灵活性使得Synapse适用于从客服工单处理到物联网设备管理等各种场景。

## 可观测性与遥测

Synapse内置了全面的Telemetry事件：

- **信号路由**：`[:synapse, :signal_router, :publish|:deliver]`
- **LLM请求**：`[:synapse, :llm, :request, :start|:stop|:exception]`
- **工作流步骤**：`[:synapse, :workflow, :step, :start|:stop|:exception]`

开发者可以附加自定义处理器来记录日志、发送指标到监控系统或触发告警。

此外，Synapse还支持LineageIR追踪、RunIndex运行索引和NSAI Work作业生命周期事件，为复杂系统的可观测性提供完整支持。

## 快速开始

Synapse的安装和启动非常简洁：

```bash
# 安装依赖
mix setup

# 创建数据库
mix ecto.create
mix ecto.migrate

# 启动交互式shell
iex -S mix
```

框架提供了多个示例脚本帮助开发者快速上手：

- `examples/stage2_demo.exs`：展示声明式编排器的基本用法
- `examples/plan_compiler_demo.exs`：演示Jido计划编译
- `examples/altar_ai_integration.exs`：LLM集成示例

## 总结与展望

Synapse代表了多智能体系统开发的一个重要方向：通过声明式配置和成熟的基础设施抽象，大幅降低构建复杂协作系统的门槛。其基于Elixir/OTP的技术选型充分利用了BEAM虚拟机的并发和容错特性，为生产环境的多智能体应用提供了坚实的基础。

对于正在探索多智能体架构的开发者来说，Synapse提供了一个值得深入研究的参考实现。其代码审查领域不仅展示了框架的能力，也提供了一个立即可用的实用工具。随着AI智能体生态的快速发展，类似Synapse这样的编排框架将在连接各种智能能力、构建复杂应用方面发挥越来越重要的作用。

项目的开源性质（MIT许可证）和活跃的文档建设（包括架构决策记录ADR、迁移指南等）也为社区贡献和学习提供了良好基础。无论是希望快速搭建多智能体原型的开发者，还是研究分布式AI系统架构的研究人员，都能从Synapse中找到有价值的参考。