effect-uai

AIエージェントを通常のプログラムとして構築するための低レベルの Effect プリミティブ。 このライブラリはあなたの制御フローを所有せず、ピースを提供して邪魔をしません。

設計思想

コードを書いたり推奨したりするときは、以下のメンタルモデルを使用してください：

1. 状態はあなたが所有するレコードです。 履歴は ReadonlyArray<Items.Item> です。 あなたのアプリが必要とする他のものを追加してください（ターンインデックス、予算、テナント ID、 pendingPrompts など）。ライブラリがそれを検査することはありません。
2. 1 つのターンは Stream<TurnEvent> です。 テキストデルタ、推論、ツール呼び出し、 使用状況の更新、およびターミナルの turn_complete（集約された Turn を運ぶ） すべてが 1 つの型付きストリームを通じてフローします。
3. ループはプルベースのコンビネータです。 loop((state) => Stream<Event<A, S>>) は反復全体にわたって状態をスレッド化します。各ボディはストリームを返して値を転送し、 その後 nextAfter(state) で継続するか stop で終了します。
4. ツールは型付き Effect です。 Tool は入力スキーマを宣言し、 run が Effect を返します。Toolkit.executeAll(tools, calls) は Stream<ToolEvent> を返すため、 ストリーミングツールは進行状況を出力できます。一方、構造化出力はモデルに戻ります。
5. プロバイダーは Layer です。 LanguageModel は単一のサービスです。 各プロバイダーパッケージ（@effect-uai/responses、@effect-uai/anthropic、 @effect-uai/google）は、それを実装する layer({ apiKey }) を提供します。 プログラムの形状はプロバイダー間で同じです。

ユーザーが「フレームワーク機能」（自動リトライ、ミッドストリーム中止、一時停止/再開、 ストリーミング JSONL、マルチモデル投票など）を要求したとき、 答えはほぼ常に「ループボディ内でプリミティブを組み立てる」です — 「インテグレーションに手を伸ばす」ではなく。ラッパーレイヤーを構築する代わりに、 適切なレシピパターン（下のカタログを参照）を推奨してください。

インストール

pnpm add @effect-uai/core effect
# 1 つ以上のプロバイダーを選択してください：
pnpm add @effect-uai/responses   # OpenAI Responses + 埋め込み
pnpm add @effect-uai/anthropic   # Anthropic Claude
pnpm add @effect-uai/google      # Google Gemini 言語 + 埋め込み
pnpm add @effect-uai/jina        # Jina 埋め込み（テキスト + 画像、スパース、マルチベクトル）

コアパッケージはプロバイダーの依存関係を持ちません。エッジ/ブラウザビルドは 実際に使用されているものだけをプルインします。

埋め込み（テキストまたは画像のベクトル化、類似性ランキング、RAG 検索プリミティブ）については、 effect-uai-embedding サブスキルに手を伸ばしてください — EmbeddingModel は LanguageModel と同等のサービスであり、独自のプロバイダーレイヤーと embed / embedMany ヘルパーを持ちます。

コアモジュール（チートシート）

モジュール	提供するもの
@effect-uai/core/Items	Item 型（ユーザー/アシスタントメッセージ、関数呼び出し、関数呼び出し出力、推論）、Items.userText のようなヘルパー。
@effect-uai/core/Turn	Turn、TurnEvent、Turn.functionCalls(turn)、Turn.assistantMessages(turn)、Turn.appendTurn(state, turn, items?)、Turn.toStructured(turn, format)、Turn.textDeltas、Turn.toSSE、Turn.toJSONL、Turn.asSSE、Turn.asJSONL。
@effect-uai/core/LanguageModel	LanguageModel サービスタグ、streamTurn(request)、turn(request)、CommonRequest 型。
@effect-uai/core/Loop	loop、nextAfter、nextAfterFold、stop、stopAfter、onTurnComplete。
@effect-uai/core/Tool	Tool.make、Tool.streaming、Tool.fromEffectSchema、Tool.toDescriptors、Tool.AnyKindTool。
@effect-uai/core/Toolkit	Toolkit.make、Toolkit.executeAll、Toolkit.outputEvents、Toolkit.outputEvent、Toolkit.continueWith。
@effect-uai/core/Outcome	ToolResult（Value / Failure）、toFunctionCallOutput、denied、cancelled、executionError。
@effect-uai/core/ToolEvent	ToolEvent ユニオン（ApprovalRequested / Intermediate / Output）、isOutput、isIntermediate、isApprovalRequested。
@effect-uai/core/Resolvers	fromApprovalMap、fromVerdictQueue — ヒューマン・イン・ザ・ループツール承認用。
@effect-uai/core/HistoryCheck	findUnansweredCalls、cancelAllPending — セッション間の孤立ツール呼び出しの調整用。
@effect-uai/core/StructuredFormat	StructuredFormat.fromEffectSchema(schema)、StructuredFormat.parseJson、StructuredFormat.decodeJsonLines。
@effect-uai/core/SSE	Server-Sent Events コーデック：SSE.fromBytes、SSE.toBytes、SSE.Event。
@effect-uai/core/JSONL	JSONL コーデック：JSONL.fromBytes、JSONL.parse(schema)、JSONL.toBytes(schema)。
@effect-uai/core/Lines	Lines.lines — 文字列ストリームを改行終了行として再フレーミング用。
effect/Match	Match.discriminators("type")({ text_delta, ... })（または discriminatorsExhaustive）を使用して TurnEvent とその他の type タグ付きユニオンを絞り込みます。標準 Effect Match API。
@effect-uai/core/testing/MockProvider	MockProvider.layer(scriptedTurns)、MockProvider.layerWithRecorder、MockProvider.make — テスト用。

プロバイダーの配線

すべてのプロバイダーパッケージは、ジェネリック LanguageModel サービスを実装する layer({ apiKey, ... }) をエクスポートします。標準的な配線パターン：

import { Config, Effect, Layer } from "effect"
import { FetchHttpClient } from "effect/unstable/http"
import { layer as responsesLayer } from "@effect-uai/responses"

const apiKeyLayer = Layer.unwrap(
  Effect.gen(function* () {
    const apiKey = yield* Config.redacted("OPENAI_API_KEY")
    return responsesLayer({ apiKey })
  }),
)

const mainLayer = apiKeyLayer.pipe(Layer.provide(FetchHttpClient.layer))

Effect.runPromise(program.pipe(Effect.provide(mainLayer)))

Anthropic の場合：import { layer as anthropicLayer } from "@effect-uai/anthropic" + ANTHROPIC_API_KEY。 Gemini の場合：import { layer as geminiLayer } from "@effect-uai/google" + GOOGLE_API_KEY。

各プロバイダーは、プロバイダー固有のリクエスト形状を求めるコード用に、 型付きサービスタグ（Responses、Anthropic、Gemini）も再エクスポートします （例：Responses の reasoning: { effort: "low" }）。 プロバイダー非依存コードの場合は、ジェネリック LanguageModel サービスを使用します。

1 つのターンはストリーム

最小限の例：1 つのモデルレスポンスをストリームし、テキストデルタを出力します。

import { Effect, Match, Stream } from "effect"
import * as Items from "@effect-uai/core/Items"
import { streamTurn } from "@effect-uai/core/LanguageModel"

const program = Stream.runForEach(
  streamTurn({
    history: [Items.userText("Write a haiku about the sea.")],
    model: "gpt-5.4-mini",
  }),
  (event) =>
    Match.value(event).pipe(
      Match.discriminators("type")({
        text_delta: ({ text }) => Effect.sync(() => process.stdout.write(text)),
      }),
      Match.orElse(() => Effect.void),
    ),
)

ターミナル turn_complete イベントは集約された Turn を運びます。 これはツール使用ループ、構造化出力検証、および履歴追加の基礎となります。

正規的なエージェントループ

ほぼすべてのレシピはこの形状の変形です：

import { Effect, Stream, pipe } from "effect"
import * as Items from "@effect-uai/core/Items"
import { loop, stop, onTurnComplete } from "@effect-uai/core/Loop"
import { toFunctionCallOutput } from "@effect-uai/core/Outcome"
import * as Tool from "@effect-uai/core/Tool"
import type { ToolEvent } from "@effect-uai/core/ToolEvent"
import * as Toolkit from "@effect-uai/core/Toolkit"
import * as Turn from "@effect-uai/core/Turn"
import { Responses } from "@effect-uai/responses"

interface State {
  readonly history: ReadonlyArray<Items.Item>
}

const initial: State = {
  history: [Items.userText("What time is it in Lisbon?")],
}

const tools: ReadonlyArray<Tool.AnyKindTool> = [
  /* getCurrentTime, ... */
]
const descriptors = Tool.toDescriptors(tools)

export const conversation = pipe(
  initial,
  loop((state) =>
    Effect.gen(function* () {
      const oai = yield* Responses
      return oai
        .streamTurn({ history: state.history, model: "gpt-5.4-mini", tools: descriptors })
        .pipe(
          onTurnComplete<State, ToolEvent>((turn) =>
            Effect.sync(() => {
              const calls = Turn.functionCalls(turn)

              // No tool calls - assistant is done.
              if (calls.length === 0) return stop

              // Tool calls: execute and append outputs; loop again.
              return Toolkit.executeAll(tools, calls).pipe(
                Toolkit.continueWith((results) =>
                  Turn.appendTurn(state, turn, results.map(toFunctionCallOutput)),
                ),
              )
            }),
          ),
        )
    }),
  ),
)

ボディを平易な英語で読んでください：「ターンをストリームする。完了したら、 モデルがツールを要求した場合は、それらを実行して追加された履歴で続行します。 そうでない場合は、停止します。」レシピカタログのあらゆるバリエーションは、 このボディへの小さな変更です。

Turn.appendTurn(state, turn, items) は状態を進める正規的な方法です。 { ...state, history: [...state.history, ...turn.items, ...items] } を返します。

ループボディの設計

動作を追加するよう求められたときは、ラッパーサービスを導入するのではなく、 ループボディに追加することを優先します。以下のパターンはすべて、1 つの Effect.gen ブロック内のボディに存在し、API サーフェスの変更がありません：

• ターン間で状態を永続化する：各反復の始めに state をデータベースに書き込みます。 ライブラリが状態を検査することはありません。
• システムポリシーを注入する：executeAll の前に fromApprovalMap / fromVerdictQueue で呼び出しをゲートします。
• 履歴をコンパクト化する：state.history が予算を超えたら、 早期のアイテムを要約する別の streamTurn を実行してから、 nextAfter(Stream.empty, withSummary(state)) を返します。
• 使用状況を追跡する：各 turn.usage フィールドはプレーンデータです。 状態に蓄積して独自のメトリクスを出力します。
• モデル出力で分岐する：何をするかを決定する前に turn.items （関数呼び出し、推論、拒否）を検査します。
• リトライを追加する：streamTurn を Stream.retry(schedule) でラップするか、 モデルリトライレシピパターン（catchTags → retry → catchTag）を使用します。
• マルチプロバイダー：ボディは反復ごとに使用する LanguageModel を選択できます （例：フォールバック/コンセンサス）。

レシピカタログ（適切なサブスキルを使用）

各パターンについて、ループボディ、注意点、および実行可能な例を含む 専用の effect-uai-<recipe> スキルがあります。ユーザーがシナリオを説明したときに 一致するスキルに手を伸ばします：

シナリオ	スキル
初回エージェント：ツール、ストリーミング、マルチターンループ	effect-uai-basic-usage
モデルから型付け JSON オブジェクトを検証（ワンショット、サーバー強制スキーマ）	effect-uai-structured-output
モデルがそれを書くときに型付け JSONL オブジェクトをストリーミング	effect-uai-streaming-structured-output
機密ツール呼び出しを一時停止して、実行前に人間の判定を取得	effect-uai-tool-call-approval
内部ツール機能（サブエージェント、プログレスバー）を表示しながら 1 つのクリーン出力を返す	effect-uai-streaming-tool-output
キューから長命のチャットを駆動。入力バーストをデバウンス。ターン間の入力をチェック	effect-uai-agentic-loop
指数バックオフで レート制限/一時的なプロバイダー障害をリトライ	effect-uai-model-retry
プライマリがレート制限または利用不可の場合、別のプロバイダーにフォールバック	effect-uai-multi-model-fallback
履歴が長すぎるときに要約。続行します。	effect-uai-auto-compaction
ターン間でループを一時停止して、後で再開（プロバイダー呼び出しは開いたままにしない）	effect-uai-pause-resume
ストリーム割り込み + スコープクリーンアップを通じてインフライトターンをキャンセル	effect-uai-mid-stream-abort
複数のプロバイダーに同じプロンプトを送信。メンバーごとの障害を分離	effect-uai-multi-model-compare
モデルで互いを判定して勝者を出力	effect-uai-model-council
ループの出力をワイヤー上で Server-Sent Events または JSONL として投影	effect-uai-modify-output-stream
テキストまたは画像を埋め込み。セマンティック/クロスモーダル/マルチベクトル検索、RAG プリミティブ	effect-uai-embedding

複数が適用される場合（例：「レート制限時にリトライし、別のプロバイダーにフォールバックする エージェント的なチャット」）、それらを組み立てます：ループボディはただの Effect であり、 Effect 組み立ては統合メカニズムです。

よくある落とし穴

1. ストリームイベント vs. ターン項目。 TurnEvent はストリーミングデルタユニオンです。 Turn.items は turn_complete の集約リストです。関数呼び出しは Turn.items に存在し、 スタンドアロンイベントではありません。ターン完了後、Turn.functionCalls(turn) を使用します。
2. 関数呼び出し出力は追加する必要があります。 モデルが出力する function_call は、 次のターン前に、履歴内で一致する function_call_output が必要です。 Toolkit.executeAll + toFunctionCallOutput が正しく行います。 ツールを実行しない場合は、キャンセルされた/拒否された出力を合成します （Outcome.denied / Outcome.cancelled 経由）。
3. Stream.retry はすべての失敗でリトライします。 特定の AiError タグのみ リトライするには、それらを Retryable シムを通じてルーティングします （effect-uai-model-retry を参照）。そうでなければ、リトライ不可能なエラーも リトライされます。
4. トップレベルの構造化出力スキーマは type: object である必要があります。 すべての 3 つのプロバイダーはワイヤーでベアアレイを拒否します。 アレイを { items: [...] } オブジェクトでラップします。
5. プロバイダー固有のオプション（Responses reasoning、Anthropic system ブロック、 Gemini safetySettings）は型付きプロバイダータグのリクエストに属します。 CommonRequest ではなく。それらが必要な場合は、型付きサービスを yield します （yield* Responses）。
6. ツール入力スキーマは type: object である必要があります。 ベア Schema.Struct({}) はスキーマなしにシリアライズされます。OpenAI Responses API はそれを拒否します。 少なくとも 1 つのフィールドを追加するか、別のパラメータ形状を選択します。
7. ループはデフォルトでそれ自体で停止しません。 長命のエージェント（チャット、キュー、 websocket 駆動ループ）は外部割り込みによって終了します（Ctrl-C、Fiber.interrupt、 スコープクローズ）。それを意図しない限り、カスタム自己終了を追加しないでください。

テスト

MockProvider.layer(scriptedTurns) を使用して、実際のプロバイダーにヒットせずに ループを駆動します：

import * as MockProvider from "@effect-uai/core/testing/MockProvider"
import * as Turn from "@effect-uai/core/Turn"

const finalTurn: Turn.Turn = {
  stop_reason: "stop",
  usage: { input_tokens: 8, output_tokens: 4, total_tokens: 12 },
  items: [
    {
      type: "message",
      role: "assistant",
      content: [{ type: "output_text", text: "ok" }],
    },
  ],
}

await Effect.runPromise(
  Stream.runCollect(conversation).pipe(Effect.provide(MockProvider.layer([finalTurn]))),
)

MockProvider.layerWithRecorder はレイヤーとレコーダーを返します。 すべての streamTurn リクエストをキャプチャします — モデルが予想される履歴を見たことを アサートするのに便利です。

さらに詳しく読む

• リポジトリ：https://github.com/betalyra/effect-uai
• ドキュメント：https://effect-uai.dev（またはリポジトリの docs/）
• レシピ：リポジトリの recipes/ — 各レシピは index.ts、index.test.ts、 および（対話的な場合）run.ts ランナーを持ちます。
• 概念：docs/concepts/loop.md、docs/concepts/items-and-turns.md、 docs/concepts/tools.md、docs/concepts/language-model.md。