重命名注释 : 这个工程以前叫做L0. 软件包/存储器正在被重新命名为可靠的AI流. 现有进口仍然得到支持。 安装 : npm install reliable-ai-streams

L0 - AI 的定型流执行底物(DSES)

所有AI流缺少的可靠性和可观察性.

LLMS在低集成运输层上产生高值推理. 流出摊位,放出活物,重排事由,违反定时保证,不暴露定时合同.

这破碎的复试。 它打破了监管。 它打破了再生产. 它使得可靠的AI系统无法在原始供应商流上再上架.

L0是固定运输的决定性执行底板——有专为流层设计的护栏:流中性,以图案为主,循环安全,有定时意识.

结果:生产级,完整性保护,决定性的AI流你终于可以建立真正的系统了.

和我们一起工作 开放AI, (中文(简体) ), 维塞尔 AI SDK, (中文(简体) ), 马斯特拉大赦国际,以及 自定义适配器支持 多式联运流,工具调用,并提供完整的决定重放.

npm install @ai2070/l0

也可在 Python 中找到: @ai-270. 可靠流派 uv add reliable-ai-streams-py - 实现全生命周期和事件签名均等.

生产等级可靠性. 只要通过你的溪流。 L0从这里接手.

L0包括涵盖所有主要可靠性特性的3,000+测试.

   Any AI Stream                    L0 Layer                         Your App
 ─────────────────    ┌──────────────────────────────────────┐    ─────────────
                      │                                      │
   Vercel AI SDK      │   Retry · Fallback · Resume          │      Reliable
   OpenAI / Mastra ──▶│   Guardrails · Timeouts · Consensus  │─────▶ Output
   Custom Streams     │   Full Observability                 │
                      │                                      │
                      └──────────────────────────────────────┘
 ─────────────────                                                ─────────────
  text / image /           L0 = Token-Level Reliability
  video / audio

即将推出的版本 :

• 1.0.0 (单位:千美元) - API 冻结

特征

知道你在做什么吗? 跳过教程

快速启动

与 Vercel AI SDK: 最小使用

import { l0, recommendedGuardrails, recommendedRetry } from "@ai2070/l0";
import { streamText } from "ai";
import { openai } from "@ai-sdk/openai";

const result = await l0({
  // Primary model stream
  stream: () =>
    streamText({
      model: openai("gpt-5-mini"),
      prompt,
    }),
});

// Read the stream
for await (const event of result.stream) {
  if (event.type === "token") process.stdout.write(event.value);
}

Vercel AI SDK: 有可靠性

import { l0, recommendedGuardrails, recommendedRetry } from "@ai2070/l0";
import { streamText } from "ai";
import { openai } from "@ai-sdk/openai";

const result = await l0({
  stream: () => streamText({ model: openai("gpt-4o"), prompt }),
  fallbackStreams: [() => streamText({ model: openai("gpt-4o-mini"), prompt })],

  // Optional: Content-agnostic, text-based guardrails
  guardrails: recommendedGuardrails,

  // Optional: Retry configuration, default as follows
  retry: {
    attempts: 3, // LLM errors only
    maxRetries: 6, // Total (LLM + network)
    baseDelay: 1000,
    maxDelay: 10000,
    backoff: "fixed-jitter", // "exponential" | "linear" | "fixed" | "full-jitter"
  },
  // Or use presets:
  // minimalRetry       // { attempts: 2, maxRetries: 4, backoff: "linear" }
  // recommendedRetry   // { attempts: 3, maxRetries: 6, backoff: "fixed-jitter" }
  // strictRetry        // { attempts: 3, maxRetries: 6, backoff: "full-jitter" }
  // exponentialRetry   // { attempts: 4, maxRetries: 8, backoff: "exponential" }

  // Optional: Timeout configuration, default as follows
  timeout: {
    initialToken: 5000, // 5s to first token
    interToken: 10000, // 10s between tokens
  },

  onError: (error, willRetry) => console.log(`Error: ${error.message}`),
});

for await (const event of result.stream) {
  if (event.type === "token") process.stdout.write(event.value);
}

另见: API.md (英语) 对所有选项, 预 编 完整示例

带有 OpenAI SDK

import OpenAI from "openai";
import { l0, openaiStream, recommendedGuardrails } from "@ai2070/l0";

const openai = new OpenAI();

const result = await l0({
  stream: openaiStream(openai, {
    model: "gpt-4o",
    messages: [{ role: "user", content: "Generate a haiku about coding" }],
  }),
  guardrails: recommendedGuardrails,
});

for await (const event of result.stream) {
  if (event.type === "token") process.stdout.write(event.value);
}

与马斯特拉AI

import { Agent } from "@mastra/core/agent";
import { l0, mastraStream, recommendedGuardrails } from "@ai2070/l0";

const agent = new Agent({
  name: "haiku-writer",
  instructions: "You are a poet who writes haikus",
  model: "openai/gpt-4o",
});

const result = await l0({
  stream: mastraStream(agent, "Generate a haiku about coding"),
  guardrails: recommendedGuardrails,
});

for await (const event of result.stream) {
  if (event.type === "token") process.stdout.write(event.value);
}

带有 Zod 的结构化输出

import { structured } from "@ai2070/l0";
import { streamText } from "ai";
import { openai } from "@ai-sdk/openai";
import { z } from "zod";

const schema = z.object({
  name: z.string(),
  age: z.number(),
  occupation: z.string(),
});

const result = await structured({
  schema,
  stream: () =>
    streamText({
      model: openai("gpt-4o-mini"),
      prompt:
        "Generate a fictional person as JSON with name, age, and occupation",
    }),
  autoCorrect: true, // Fix trailing commas, missing braces, markdown fences
});

console.log(result.data); // { name: "Alice", age: 32, occupation: "Engineer" }

生命周期事件

import { l0 } from "@ai2070/l0";
import { streamText } from "ai";
import { openai } from "@ai-sdk/openai";

const result = await l0({
  stream: () => streamText({ model: openai("gpt-4o"), prompt }),

  onEvent: (event) => {
    if (event.type === "token") process.stdout.write(event.value || "");
    if (event.type === "error") console.error("Error:", event.error);
    if (event.type === "complete") console.log("\nDone!");
  },
});

for await (const _ of result.stream) {
  // Events already handled by onEvent
}

倒置模型提供商( S)

import { l0 } from "@ai2070/l0";
import { streamText } from "ai";
import { openai } from "@ai-sdk/openai";
import { anthropic } from "@ai-sdk/anthropic";

const result = await l0({
  // Primary model
  stream: () => streamText({ model: openai("gpt-4o"), prompt }),

  // Fallbacks: tried in order if primary fails (supports both model and provider fallbacks)
  fallbackStreams: [
    () => streamText({ model: openai("gpt-4o-mini"), prompt }),
    () => streamText({ model: anthropic("claude-sonnet-4-20250514"), prompt }),
  ],

  onFallback: (index, reason) => console.log(`Switched to fallback ${index}`),
});

并行执行

import { parallel } from "@ai2070/l0";
import { streamText } from "ai";
import { openai } from "@ai-sdk/openai";

const prompts = ["Name a fruit", "Name a color", "Name an animal"];

const results = await parallel(
  prompts.map((prompt) => ({
    stream: () => streamText({ model: openai("gpt-4o-mini"), prompt }),
  })),
  { concurrency: 3 },
);

results.results.forEach((r, i) => {
  console.log(`${prompts[i]}: ${r?.state.content.trim()}`);
});

管道:流管

import { pipe } from "@ai2070/l0";
import { streamText } from "ai";
import { openai } from "@ai-sdk/openai";

const result = await pipe(
  [
    {
      name: "summarize",
      fn: (input) => ({
        stream: () =>
          streamText({
            model: openai("gpt-4o"),
            prompt: `Summarize: ${input}`,
          }),
      }),
    },
    {
      name: "translate",
      fn: (summary) => ({
        stream: () =>
          streamText({
            model: openai("gpt-4o"),
            prompt: `Translate to French: ${summary}`,
          }),
      }),
    },
  ],
  longDocument,
);

console.log(result.output); // French translation of summary

哲学

• 没有魔法 - 一切都是明确和可预测的
• 流动第一 - 为实时送信而建造
• 信号,不是重写 - 护栏检测问题,不要修改输出
• 模型不可知论 - 使用任何模型
• 零依赖关系 - Vercel AI SDK、OpenAI SDK或Mastra AI只有(可选的)同伴依赖

套装大小( 最小) :

无依无依. 树能震动的子路径出口 最小捆绑.

大多数应用程序只应使用 import { l0 } from "@ai2070/l0"。 。 。 . 只有针对边缘运行时间或严格的捆绑限制时,才能优化进口.

Zod 验证程序Schemas

L0 输出 Zod 4 计程器用于所有 L0 类型的运行时间验证 :

import {
  L0StateSchema,
  L0EventSchema,
  GuardrailViolationSchema,
} from "@ai2070/l0/zod";

// Validate runtime data
const state = L0StateSchema.parse(unknownData);

// Type-safe validation
const result = L0EventSchema.safeParse(event);
if (result.success) {
  console.log(result.data.type);
}

所有核心类型都可以使用图表: L0State, (中文(简体) ), L0Event, (中文(简体) ), L0Telemetry, (中文(简体) ), RetryOptions, (中文(简体) ), GuardrailViolation, (中文(简体) ), ConsensusResult, (中文(简体) ), PipelineResult还有更多.

业绩

Apple M1 Max, Node.js 24, 零延迟模拟流的基准(2000个令牌):

全堆 = JSON + Markdown + 零输出护栏 + 漂移检测 + 检查. 见 银行 详细情况.

文档

支助

L0是独立开发和维护的. 如果您的公司依赖于L0或想要支持正在进行的开发(包括Python版本,网站docs,以及未来的工具),可以自由接触:

制造商[email protected]

许可证

阿帕奇 - 2.0