iii-architecture-patterns

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

Architecture Patterns

架构模式

Use this skill when the task is larger than one function or trigger. Pick the pattern from the requirement, then implement it with

Function

Trigger

Worker

, state, queues, streams, and pub/sub.

当任务规模大于单个函数或触发器时，使用此技能。从需求中选择合适的模式，然后通过

Function

、

Trigger

、

Worker

、状态、队列、流以及发布/订阅来实现。

Pattern Map

模式映射表

Requirement	Pattern	iii shape
Sequential work with retries, DLQ, step tracking	Durable workflow	Functions chained through named queues, progress in state
Keep views, metrics, cache, or clients in sync	Reactive backend	State triggers plus stream/pubsub side effects
Specialized AI agents hand work to each other	Agentic backend	One function per agent, queue handoffs, shared state
Commands publish events and projections update independently	Event-driven CQRS	Command functions, event log in state, subscribe triggers
Pure, traceable composition	Effect pipeline	Small functions composed synchronously with `trigger()`
Webhook/cron automation chains	Low-code automation	Trigger, transform, action nodes chained by enqueue

需求	模式	iii 结构
带有重试、死信队列（DLQ）、步骤跟踪的顺序工作	持久化工作流	通过命名队列串联的函数，状态中记录进度
保持视图、指标、缓存或客户端同步	响应式后端	状态触发器加上流/发布订阅的副作用
专业AI代理之间协作处理工作	智能代理后端	每个代理对应一个函数，通过队列交接工作，共享状态
命令发布事件，投影独立更新	事件驱动型CQRS	命令函数，状态中存储事件日志，订阅触发器
纯函数式、可追踪的组合	效果流水线	小型函数通过 `trigger()` 同步组合
Webhook/定时任务自动化链	低代码自动化	触发、转换、动作节点通过入队操作串联

Durable Workflow

持久化工作流

Use named queues when steps need different retry, FIFO, or concurrency policy.

当步骤需要不同的重试、FIFO或并发策略时，使用命名队列。

TypeScript

typescript

import { registerWorker, TriggerAction } from "iii-sdk";

const iii = registerWorker("ws://localhost:49134", { workerName: "order-workflow" });

async function track(orderId: string, step: string, status: string) {
  await iii.trigger({
    function_id: "state::update",
    payload: { scope: "orders", key: orderId, ops: [{ op: "set", path: `/steps/${step}`, value: status }] },
  });
}

iii.registerFunction("orders::validate", async (order) => {
  await track(order.id, "validate", "done");
  return iii.trigger({
    function_id: "orders::charge",
    payload: order,
    action: TriggerAction.Enqueue({ queue: "order-payment" }),
  });
});

iii.registerFunction("orders::charge", async (order) => {
  await track(order.id, "payment", "done");
  return iii.trigger({
    function_id: "orders::ship",
    payload: order,
    action: TriggerAction.Enqueue({ queue: "order-ship" }),
  });
});

typescript

import { registerWorker, TriggerAction } from "iii-sdk";

const iii = registerWorker("ws://localhost:49134", { workerName: "order-workflow" });

async function track(orderId: string, step: string, status: string) {
  await iii.trigger({
    function_id: "state::update",
    payload: { scope: "orders", key: orderId, ops: [{ op: "set", path: `/steps/${step}`, value: status }] },
  });
}

iii.registerFunction("orders::validate", async (order) => {
  await track(order.id, "validate", "done");
  return iii.trigger({
    function_id: "orders::charge",
    payload: order,
    action: TriggerAction.Enqueue({ queue: "order-payment" }),
  });
});

iii.registerFunction("orders::charge", async (order) => {
  await track(order.id, "payment", "done");
  return iii.trigger({
    function_id: "orders::ship",
    payload: order,
    action: TriggerAction.Enqueue({ queue: "order-ship" }),
  });
});

Python

python

from iii import register_worker

iii = register_worker("ws://localhost:49134")

def track(order_id, step, status):
    iii.trigger({
        "function_id": "state::update",
        "payload": {
            "scope": "orders",
            "key": order_id,
            "ops": [{"op": "set", "path": f"/steps/{step}", "value": status}],
        },
    })

def validate(order):
    track(order["id"], "validate", "done")
    return iii.trigger({
        "function_id": "orders::charge",
        "payload": order,
        "action": {"type": "enqueue", "queue": "order-payment"},
    })

iii.register_function("orders::validate", validate)

python

from iii import register_worker

iii = register_worker("ws://localhost:49134")

def track(order_id, step, status):
    iii.trigger({
        "function_id": "state::update",
        "payload": {
            "scope": "orders",
            "key": order_id,
            "ops": [{"op": "set", "path": f"/steps/{step}", "value": status}],
        },
    })

def validate(order):
    track(order["id"], "validate", "done")
    return iii.trigger({
        "function_id": "orders::charge",
        "payload": order,
        "action": {"type": "enqueue", "queue": "order-payment"},
    })

iii.register_function("orders::validate", validate)

Rust

rust

use iii_sdk::{TriggerAction, TriggerRequest};
use serde_json::json;

async fn enqueue_charge(iii: iii_sdk::III, order: serde_json::Value) -> Result<serde_json::Value, iii_sdk::IIIError> {
    iii.trigger(TriggerRequest {
        function_id: "state::update".into(),
        payload: json!({
            "scope": "orders",
            "key": order["id"],
            "ops": [{ "op": "set", "path": "/steps/validate", "value": "done" }]
        }),
        action: None,
        timeout_ms: None,
    }).await?;

    iii.trigger(TriggerRequest {
        function_id: "orders::charge".into(),
        payload: order,
        action: Some(TriggerAction::Enqueue { queue: "order-payment".into() }),
        timeout_ms: None,
    }).await
}

rust

use iii_sdk::{TriggerAction, TriggerRequest};
use serde_json::json;

async fn enqueue_charge(iii: iii_sdk::III, order: serde_json::Value) -> Result<serde_json::Value, iii_sdk::IIIError> {
    iii.trigger(TriggerRequest {
        function_id: "state::update".into(),
        payload: json!({
            "scope": "orders",
            "key": order["id"],
            "ops": [{ "op": "set", "path": "/steps/validate", "value": "done" }]
        }),
        action: None,
        timeout_ms: None,
    }).await?;

    iii.trigger(TriggerRequest {
        function_id: "orders::charge".into(),
        payload: order,
        action: Some(TriggerAction::Enqueue { queue: "order-payment".into() }),
        timeout_ms: None,
    }).await
}

Reactive Backend

响应式后端

Use state triggers when the requirement says "after create/update, do this", "avoid polling", or "push live updates".

typescript

iii.registerFunction("todos::on-change", async (event) => {
  await iii.trigger({
    function_id: "stream::send",
    payload: { stream_name: "todos-live", group_id: "default", data: event.new_value },
    action: TriggerAction.Void(),
  });
});

iii.registerTrigger({
  type: "state",
  function_id: "todos::on-change",
  config: { scope: "todos" },
});

当需求提到“创建/更新后执行此操作”、“避免轮询”或“推送实时更新”时，使用状态触发器。

typescript

iii.registerFunction("todos::on-change", async (event) => {
  await iii.trigger({
    function_id: "stream::send",
    payload: { stream_name: "todos-live", group_id: "default", data: event.new_value },
    action: TriggerAction.Void(),
  });
});

iii.registerTrigger({
  type: "state",
  function_id: "todos::on-change",
  config: { scope: "todos" },
});

Agentic Backend

智能代理后端

Model each agent as a function with one responsibility. Store shared context in state and hand off work through named queues.

typescript

iii.registerFunction("agents::researcher", async (task) => {
  await iii.trigger({
    function_id: "state::set",
    payload: { scope: "research", key: task.id, value: { findings: [] } },
  });
  return iii.trigger({
    function_id: "agents::critic",
    payload: task,
    action: TriggerAction.Enqueue({ queue: "agent-tasks" }),
  });
});

将每个代理建模为单一职责的函数。在状态中存储共享上下文，并通过命名队列交接工作。

typescript

iii.registerFunction("agents::researcher", async (task) => {
  await iii.trigger({
    function_id: "state::set",
    payload: { scope: "research", key: task.id, value: { findings: [] } },
  });
  return iii.trigger({
    function_id: "agents::critic",
    payload: task,
    action: TriggerAction.Enqueue({ queue: "agent-tasks" }),
  });
});

Event-Driven CQRS

事件驱动型CQRS

Commands validate and publish domain events. Projections subscribe independently and write query-optimized state.

typescript

iii.registerFunction("cmd::add-inventory-item", async (input) => {
  const event = { type: "inventory.item-added", itemId: input.itemId, quantity: input.quantity };
  await iii.trigger({
    function_id: "state::set",
    payload: { scope: "inventory-events", key: `${Date.now()}-${input.itemId}`, value: event },
  });
  await iii.trigger({ function_id: "publish", payload: { topic: event.type, data: event } });
  return { accepted: true };
});

iii.registerTrigger({
  type: "subscribe",
  function_id: "proj::inventory-list",
  config: { topic: "inventory.item-added" },
});

命令负责验证并发布领域事件。投影独立订阅事件，并写入优化查询的状态。

typescript

iii.registerFunction("cmd::add-inventory-item", async (input) => {
  const event = { type: "inventory.item-added", itemId: input.itemId, quantity: input.quantity };
  await iii.trigger({
    function_id: "state::set",
    payload: { scope: "inventory-events", key: `${Date.now()}-${input.itemId}`, value: event },
  });
  await iii.trigger({ function_id: "publish", payload: { topic: event.type, data: event } });
  return { accepted: true };
});

iii.registerTrigger({
  type: "subscribe",
  function_id: "proj::inventory-list",
  config: { topic: "inventory.item-added" },
});

Selection Rules

选择规则

Use sync composition for short effect pipelines where the caller needs the final result.
Use enqueue for unreliable, slow, or must-complete steps.
Use pub/sub for independent fan-out where each subscriber can tolerate event-style delivery.
Use state triggers for derived views and side effects that should run automatically after writes.
Use cron triggers for scheduled maintenance and periodic starts.
Keep each function small enough to test independently.

对于调用方需要最终结果的短效果流水线，使用同步组合。
对于不可靠、缓慢或必须完成的步骤，使用入队操作。
对于独立扇出且每个订阅者可容忍事件式交付的场景，使用发布/订阅。
对于派生视图以及写入后应自动运行的副作用，使用状态触发器。
对于计划维护和周期性启动任务，使用定时触发器。
保持每个函数足够小，以便独立测试。

When to Use

使用场景

Use this skill for complete backend patterns: workflows, agentic systems, reactive apps, CQRS, effect pipelines, and automation chains.
Use this when the request describes product behavior rather than a single SDK API call.

当需要构建完整的后端模式时使用此技能：工作流、智能代理系统、响应式应用、CQRS、效果流水线以及自动化链。
当请求描述的是产品行为而非单个SDK API调用时使用此技能。

Boundaries

边界说明

For exact trigger config, function registration syntax, custom triggers, channels, and HTTP-invoked functions, use
```
iii-core-primitives
```
.
For queue retry, FIFO, adapter, port, and worker-manager config, use
```
iii-engine-config
```
.
For package-specific SDK syntax, use
```
iii-sdk-reference
```
.
Worker-backed capability details live with the worker docs, not as top-level iii skills.

若需精确的触发器配置、函数注册语法、自定义触发器、通道以及HTTP调用函数，请使用
```
iii-core-primitives
```
。
若需队列重试、FIFO、适配器、端口以及Worker管理器配置，请使用
```
iii-engine-config
```
。
若需特定包的SDK语法，请使用
```
iii-sdk-reference
```
。
Worker相关的能力细节请参考Worker文档，而非顶层iii技能文档。