使用 Erlang/Elixir 与 OTP 构建智能体系统：旧模式与新热潮

作者：Matheus de Camargo Marques

引言：热潮与模式

每隔几个月，AI 领域就会有人发布一个新的框架，用于运行自主智能体（AI Agent）。这个框架承诺：隔离状态、消息传递、监督层级和故障恢复。BEAM 社区的人看了，点点头，然后回去继续干活。

这种现象之所以不断发生，是因为基于进程的并发确实解决了一个真正困难的问题，而 BEAM 虚拟机自 1986 年以来就一直在解决这个问题——不是作为一个库，不是作为一种你选择采用的模式，而是作为运行时本身。

Python AI 生态正在构建的智能体框架，正独立地收敛到相同的架构：隔离进程、消息传递、监督层级、故障恢复。这些模式并非偶然与 OTP 相似。它们相似是因为问题本身就需要这种形态。

在本文中，我想展示我们在 Erlang/OTP 中经过四十年打磨的模式——GenServer、Process、Supervisor——如何几乎一对一地映射到行业现在正在"发现"的智能体模式。而且我想用可运行的 Elixir 代码来展示，代码片段足够小，可以截图分享。

第一部分：从微服务看智能体模式（以及它们为何已经是 OTP）

那张人人都在发的映射表

2026 年 4 月，Teksystems 发布了一篇文章，将微服务概念直接映射到对应的智能体 AI 概念：

让我感到震撼的不是有什么新东西。而是有多少东西早已存在。这张表中的每一个"智能体模式"都能在 OTP 中找到直接对应：

• 服务发现 → OTP 中的 Registry 和 :global
• API 网关 / Supervisor → Supervisor 和 DynamicSupervisor
• 状态管理 → GenServer state 和 :ets 表
• 结构化契约 → 模式匹配和消息协议

整个行业正在重新发明 BEAM 几十年来作为原生运行时特性已有的东西。

"认知单体"问题

VentureBeat 在 2025 年 12 月完美地抓住了核心问题："如果你试图把 20 页的系统指令塞进一次大语言模型调用来构建企业级 AI，你正在重蹈一个熟悉的架构错误。你不是在构建一个智能体——你是在构建一个认知单体。"

这与我们早在 2010 年代初针对单体架构识别出的失败模式完全相同。一个全知全能的单一进程成为瓶颈。它难以专业化、造成单点故障，而且无法管理复杂的带状态工作流。

解决方案，一如既往，是分解。而 OTP 中分解的单元就是进程。

第二部分：为什么 BEAM/OTP 是智能体一直在等待的运行时

并发问题，简要陈述

你的程序需要同时做多件事。你有数千个并发对话，每个都有自己的状态。你需要隔离故障，让一个坏的智能体不会拖垮整个系统。你需要从崩溃中优雅地恢复。

有两种根本方法：

带锁的共享状态。 多个线程访问同一块内存。你用互斥锁、信号量和锁来防止数据损坏。问题不在于它不能工作——而是在于它"能工作"直到"不能工作"的那一刻。

带消息传递的隔离状态。 每个并发单元有自己的内存。通信的唯一方式是通过发送消息。没有共享内存，没有锁，没有竞态。这就是 Actor 模型。Carl Hewitt 在 1973 年提出了它。Erlang 在 1986 年将其实现为运行时。

每隔几年，行业其余部分就会重新发现它。

"让它崩溃"哲学不是鲁莽

Zylos Research 关于 AI 智能体监督树的文章说得非常好："构建有弹性的 AI 智能体运行时，需要 Erlang 工程师在 1980 年代对电信系统应用的相同纪律：接受进程会失败，隔离爆炸半径，自动化恢复"。

"让它崩溃"的哲学不是关于粗心大意。它是将故障处理逻辑与业务逻辑分离。一个崩溃后被 supervisor 重启的 GenServer，要比试图内联处理所有可能错误的进程更可靠。

第三部分：代码示例——用 Elixir 构建智能体系统

示例 1：一个简单的 GenServer 智能体

让我们从基础开始。一个基于 GenServer 的智能体，持有状态并能处理消息。

Snippet 1 — 模块结构和公共 API：

defmodule MyApp.ResearchAgent do
  use GenServer

  def start_link(opts \\ []) do
    name = Keyword.get(opts, :name, __MODULE__)
    GenServer.start_link(__MODULE__, opts, name: name)
  end

  def query(agent, prompt) do
    GenServer.call(agent, {:query, prompt}, 30_000)
  end

Snippet 2 — 初始化智能体状态：

  @impl true
  def init(opts) do
    state = %{
      history: [],
      tools: Keyword.get(opts, :tools, []),
      model: Keyword.get(opts, :model, "claude-sonnet-4"),
      max_history: Keyword.get(opts, :max_history, 50)
    }

    {:ok, state}
  end

Snippet 3 — 主查询循环：

  @impl true
  def handle_call({:query, prompt}, _from, state) do
    context = build_context(state.history, prompt)
    response = call_llm(context, state.model)

    new_history = [
      %{role: "user", content: prompt},
      %{role: "assistant", content: response}
    ] ++ state.history

Snippet 4 — 裁剪历史记录以防止上下文稀释：

    trimmed_history = Enum.take(new_history, state.max_history)
    new_state = %{state | history: trimmed_history}

    {:reply, {:ok, response}, new_state}
  end

Snippet 5 — 私有辅助函数：

  defp build_context(history, prompt) do
    messages = Enum.reverse(history)
    messages ++ [%{role: "user", content: prompt}]
  end

  defp call_llm(_messages, _model) do
    "Simulated response from #{_model}"
  end
end

这是一个基本的构建块。它持有状态、处理请求、返回响应。但仅凭它自己是脆弱的。如果它崩溃了，一切都丢失了。这就是 supervisor 登场的地方。

示例 2：多智能体的监督树

OTP 的核心优势是监督树。Supervisor 监控其子进程并在故障时重启它们。

Snippet 6 — Supervisor 启动：

defmodule MyApp.AgentSupervisor do
  use Supervisor

  def start_link(opts) do
    Supervisor.start_link(__MODULE__, opts, name: __MODULE__)
  end

  @impl true
  def init(_opts) do

Snippet 7 — 用于智能体发现的 Registry：

    children = [
      {Registry, keys: :unique, name: MyApp.AgentRegistry},
      {DynamicSupervisor, name: MyApp.AgentDynamicSupervisor,
       strategy: :one_for_one},

Snippet 8 — 持久化智能体和重启策略：

      %{id: :research_agent,
        start: {MyApp.ResearchAgent, :start_link, [[name: :research_agent]]}},
      %{id: :code_agent,
        start: {MyApp.CodeAgent, :start_link, [[name: :code_agent]]}}
    ]

    Supervisor.init(children, strategy: :one_for_one,
                    max_restarts: 5, max_seconds: 30)
  end
end

这个监督树给了你一些没有任何 Python 智能体框架开箱即有的东西：自动的、隔离的恢复。如果 ResearchAgent 崩溃了，supervisor 会重启它。CodeAgent 不受影响。

示例 3：DynamicSupervisor 用于智能体池

在生产环境中，你不需要固定数量的智能体。你需要按需派生智能体。

Snippet 9 — 智能体池 API：

defmodule MyApp.AgentPool do
  use GenServer

  def start_link(opts \\ []) do
    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
  end

  def spawn_agent(agent_module, opts \\ []) do
    GenServer.call(__MODULE__, {:spawn_agent, agent_module, opts})
  end

Snippet 10 — 动态智能体派生：

  @impl true
  def handle_call({:spawn_agent, agent_module, opts}, _from, state) do
    spec = {agent_module, :start_link, [opts]}

    case DynamicSupervisor.start_child(MyApp.AgentDynamicSupervisor, spec) do
      {:ok, pid} ->
        agent_id = make_ref()
        new_agents = Map.put(state.agents, agent_id, pid)
        {:reply, {:ok, agent_id, pid}, %{state | agents: new_agents}}
    end
  end
end

这个模式直接对应了 VentureBeat 所说的"对话蜂群"——多个智能体协作。在 OTP 中，它只是一个 DynamicSupervisor 管理着一组 GenServer 进程。

示例 4：gen_statem 用于 ReAct 循环

对于智能体系统来说最有趣的模式是 ReAct 循环（推理 + 行动）。大多数开发者会想到 GenServer。但 gen_statem 将复杂循环转化为声明式状态机。

Snippet 11 — ReAct 智能体定义：

defmodule MyApp.ReActAgent do
  @behaviour :gen_statem

  def start_link(opts \\ []) do
    :gen_statem.start_link({:local, __MODULE__}, __MODULE__, opts, [])
  end

  def run(task) do
    :gen_statem.call(__MODULE__, {:run, task}, 60_000)
  end

Snippet 12 — 初始化状态机状态：

  @impl true
  def callback_mode, do: :state_functions

Snippet 13 — 从 idle 到 reasoning 的状态转换：

def idle({:call, from}, {:run, task}, state) do new_state = %{state | task: task, history: [%{role: "user", content: task}], iteration: 0}

{:next_state, :reasoning, new_state, [{:reply, from, :started}]} end

Snippet 14 — Reasoning 状态（LLM 调用）：

def reasoning(:internal, _data, state) do response = call_llm(state.history, state.tools, state.model)

case parse_response(response) do {:action, tool_name, tool_input} -> new_history = [%{role: "assistant", content: response} | state.history] {:next_state, :acting, %{state | history: new_history}, [{:next_event, :internal, {:execute_tool, tool_name, tool_input}}]}

Snippet 15 — Reasoning 状态结束处理：

  {:final, answer} ->
    new_history = [%{role: "assistant", content: answer} | state.history]
    {:next_state, :done, %{state | history: new_history},
     [{:next_event, :internal, {:final_answer, answer}}]}
end

end

Snippet 16 — Acting 状态（工具执行）：

def acting(:internal, {:execute_tool, tool_name, tool_input}, state) do tool = Enum.find(state.tools, fn t -> t.name == tool_name end)

result = case tool do nil -> "Unknown tool: #{tool_name}" tool -> tool.execute.(tool_input) end

new_history = [%{role: "tool", content: result} | state.history] {:next_state, :reasoning, %{state | history: new_history}} end end

这正是 gen_statem 大放异彩的地方。智能体循环是显式的。你能清楚地看到智能体处于什么状态，以及可能发生哪些状态转换。

Example 5: 基于 ETS 的工具注册中心

在 AI 智能体系统中，智能体需要发现和调用工具。这正是服务发现模式。

Snippet 17 — 工具注册中心 API：

defmodule MyApp.ToolRegistry do use GenServer @table MODULE

def start_link(_opts) do GenServer.start_link(MODULE, [], name: MODULE) end

def register(name, description, schema, executor) do GenServer.call(MODULE, {:register, name, description, schema, executor}) end

Snippet 18 — 工具查询：

def lookup(name) do case :ets.lookup(@table, name) do [{^name, description, schema, executor}] -> {:ok, %{name: name, description: description, input_schema: schema, executor: executor}} [] -> {:error, :not_found} end end

Snippet 19 — 初始化与注册：

@impl true def init(_) do :ets.new(@table, [:named_table, :public, :set]) {:ok, %{}} end

@impl true def handle_call({:register, name, desc, schema, exec}, _from, state) do :ets.insert(@table, {name, desc, schema, exec}) {:reply, :ok, state} end end

Snippet 20 — 带错误处理的工具执行器：

defmodule MyApp.ToolExecutor do def execute(tool_name, input) do case MyApp.ToolRegistry.lookup(tool_name) do {:ok, tool} -> try do {:ok, tool.executor.(input)} rescue e -> {:error, Exception.message(e)} end {:error, :not_found} -> {:error, "Tool #{tool_name} not found"} end end end

这就是 Model Context Protocol (MCP) 模式，用 50 行 Elixir 实现。不依赖外部库。不需要 Python 运行时。不需要框架。

Example 6: GenServer 作为 LLM Harness

Snippet 21 — Harness API：

defmodule MyApp.LLMHarness do use GenServer

def start_link(opts) do GenServer.start_link(MODULE, opts, name: opts[:name]) end

def complete(harness, prompt, opts \ []) do GenServer.call(harness, {:complete, prompt, opts}, 60_000) end

Snippet 22 — Harness 状态：

@impl true def init(opts) do state = %{ provider: opts[:provider] || :anthropic, model: opts[:model] || "claude-sonnet-4", api_key: opts[:api_key] || System.get_env("ANTHROPIC_API_KEY"), tools: opts[:tools] || [] }

{:ok, state} end

Snippet 23 — Completion 循环：

@impl true def handle_call({:complete, prompt, opts}, _from, state) do messages = build_messages(prompt, opts)

case call_provider(state.provider, state.model, messages, state.tools, state.api_key) do {:ok, response} -> {:reply, {:ok, response}, state} {:error, reason} -> {:reply, {:error, reason}, state} end end

Snippet 24 — 向上游服务发起 HTTP 调用：

defp call_provider(:anthropic, model, messages, tools, api_key) do body = %{model: model, max_tokens: 4096, messages: messages, tools: Enum.map(tools, &tool_to_schema/1)}

headers = [{"x-api-key", api_key}, {"anthropic-version", "2023-06-01"}, {"content-type", "application/json"}]

Req.post("https://api.anthropic.com/v1/messages", json: body, headers: headers) end end

PiEx、Alloy 和 Omni Agent 使用的正是这一模式——它们都是在受监督的 GenServer 内部实现智能体循环的 Elixir 库。

Part IV: 免费获得的能力

Jido 框架文档说得明白："每个智能体运行在 OTP 监督下的独立 BEAM 进程中。崩溃自动恢复。如果某个智能体失败，不影响其他任何智能体"。

重启后的状态恢复

这是 AI 智能体系统中最难的问题。在 OTP 中，你有几种选择：

Checkpointing：定期将状态持久化到 :ets、:dets 或数据库。

Event sourcing：每个状态变更都是一条事件。重启时重放事件。

Snapshot + replay：定期快照，加上自上次快照以来的事件重放。

GenServer 有 terminate/2 回调，你可以在崩溃前持久化状态。

进程隔离与沙箱

PtcRunner 在 BEAM 原生沙箱中运行生成的代码，具备进程隔离、超时、堆内存限制和受控的工具访问。这不是容器。它是一个拥有独立堆的轻量级进程。

:observer 给你进程树的实时视图。:sys.get_state/1 让你检查任意 GenServer 的状态。:sys.trace/3 让你追踪消息流。

Part V: 永不过时的模式

我在 AWS Summit London 2026 引用 Matheus Guimaraes 的话："模式永不死。只是换上了新面具"。

2026 年业界正在"发现"的 AI 智能体模式并不新鲜。它们和 Erlang/OTP 自 1986 年以来所体现的模式如出一辙：

单一职责 → 每个智能体是一个独立进程。

服务发现 → Registry、:global 和 :ets 表。

API Gateway → Supervisor 和 DynamicSupervisor。

结构化契约 → 模式匹配和消息协议。

容错 → 可配置重启策略的监督树。

进程隔离 → 拥有独立堆的 BEAM 轻量级进程。

正如 Niko Maroulis 在 LinkedIn 上写的："我们的系统越'智能化'，Elixir/Erlang/OTP 就越像是为它量身打造的运行时"。

Part VI: 入门指南

从 GenServer 开始。构建一个单独的智能体。保持简单。测试循环。

添加 Supervisor。将你的智能体包裹在监督树中。

使用 DynamicSupervisor 处理池。按需动态生成智能体。

对复杂循环探索 gen_statem。让状态变得显式。

用 ETS 做工具发现。在 :ets 中注册工具。

加入可观测性。使用 :observer 和 telemetry。

看看这些库：Jido、PiEx、Alloy、Omni Agent 和 Kyber-BEAM。

结语: 旧运行时，新 hype

围绕 AI 智能体的 hype 是真实的。但模式不是。这些模式我们已打磨了数十年。

如果你是 Elixir 开发者，你已经拥有了业界正在努力构建的运行时。你不需要新框架。你只需要把已有的知识应用到新一类的工作负载上。

BEAM 就是为并发、分布式、容错、有状态而生的。这些正是 AI 智能体系统所需的特性。

Hype 会过去。模式会留下来。

Variant Systems: "BEAM OTP: Why Everyone Keeps Reinventing It" — 2026年2月22日

VentureBeat: "AI teams: 企业自动化的新蓝图" — 2025年12月15日

bombadil-labs/kyber — 基于 Elixir/OTP 构建的 LLM 智能体工具集

nshkrdotcom/jido — Elixir 自治智能体框架

nshkrdotcom/mabeam — 基于 GenServer 和监督机制的智能体框架

PiEx — Elixir AI 编程智能体库

Alloy — Elixir 通用的模型无关智能体工具集

Omni Agent — Elixir 有状态 LLM 智能体

Erlang/OTP Design Principles — Supervisor Behaviour

"Beyond GenServers: Declarative AI Flows With gen_statem" — CodeBEAM Europe 2026

Matheus de Camargo Marques 是一位专注于 Elixir、Erlang 和分布式系统的软件工程师。本文仅代表基于独立研究的个人分析，不代表任何组织的官方立场。