diff --git a/docs/en/reference/orchestration-et-supervision.md b/docs/en/reference/orchestration-et-supervision.md new file mode 100644 index 00000000..50098eea --- /dev/null +++ b/docs/en/reference/orchestration-et-supervision.md @@ -0,0 +1,36 @@ + +# Orchestrate several models and monitor calls + +The `@ai-swiss/base-llm` package exposes a single model port: every model offers the same `complete` interface. Because meta-models respect this port, they compose: each wraps one or more models and is itself a model. It therefore drops in anywhere a single model is used, from settings to routing. Choosing a model stays an explicit decision. + +## Mixture of agents + +`createMoaModel` queries several proposers in parallel, then an aggregator synthesizes their drafts into one answer. Proposers work without tools and produce text; the aggregator receives the drafts as private guidance and keeps the original tools. Token usage sums across all calls, and synthesis continues even if some proposers fail. + +```mermaid +flowchart TD + R[Request] --> P1[Proposer 1] + R --> P2[Proposer 2] + P1 --> D[Drafts] + P2 --> D + D --> A[Aggregator] + A --> O[Final answer] +``` + +## Triumvirat + +`createTriumviratModel` follows the architecture of Sakana Fugu and TRINITY. Each turn, a coordinator picks a model from a swappable pool and assigns it a role: the thinker plans, the worker produces or fixes the answer and alone receives the tools, the verifier judges the draft and decides when to stop. The loop runs until acceptance or a turn budget. + +The default coordinator relies on a model from the pool, with a deterministic thinker, worker, verifier fallback. A caller-supplied `decide` function replaces that coordinator without changing the interface. + +## Configure ensembles from settings + +An `ensembles` block in `.ai/studio.settings.json` makes these compositions available without code. Each entry names a `type` (`moa` or `triumvirat`) and member references in the `/` form. The ensemble name then works anywhere a model reference works, for example in `routing.model`. Bad configuration fails with a clear message. + +## Monitoring with Langfuse + +`createLangfuseModel` wraps any model and traces every call to Langfuse: input, output, tokens, latency, and errors. The wrapper adds no dependency: it writes to the public ingestion interface through `fetch`. The send runs in the background and adds no latency to the call; `flush()` drains in-flight sends before a short process ends. A monitoring failure never interrupts the wrapped call. Keys come from the environment variables `LANGFUSE_PUBLIC_KEY` and `LANGFUSE_SECRET_KEY`, and a self-hosted host is declared through `LANGFUSE_HOST`. + +## Choosing between mixture and triumvirat + +Mixture aims for breadth in a single turn: parallel proposals then a synthesis, simple and economical. Triumvirat aims for depth in sequence: plan, produce, verify, repeat, with self-correction and tool use. Both share the same port, so Langfuse monitoring wraps them the same way. diff --git a/docs/reference/orchestration-et-supervision.md b/docs/reference/orchestration-et-supervision.md new file mode 100644 index 00000000..c3027d43 --- /dev/null +++ b/docs/reference/orchestration-et-supervision.md @@ -0,0 +1,47 @@ +--- +schema_version: base.resource.v1 +id: orchestration-supervision +type: document +title: Orchestrer plusieurs modèles et superviser les appels +description: Composer des modèles (mixture d'agents, coordination par rôles), les configurer depuis les réglages, et tracer chaque appel vers Langfuse, le tout via le même port de modèle. +scope: public +status: active +sensitivity: public +keywords: [orchestration, supervision, observabilité, mixture, agents, langfuse, routage, modèles] +--- + +# Orchestrer plusieurs modèles et superviser les appels + +Le paquet `@ai-swiss/base-llm` expose un port de modèle unique: tout modèle offre la même interface `complete`. Comme les méta-modèles respectent ce port, ils se composent: chacun enveloppe un ou plusieurs modèles et reste lui-même un modèle. Il s'insère donc partout où un modèle simple s'utilise, des réglages au routage. Choisir un modèle reste toujours une décision explicite. + +## Mixture d'agents + +`createMoaModel` interroge plusieurs proposeurs en parallèle, puis un agrégateur synthétise leurs brouillons en une seule réponse. Les proposeurs travaillent sans outils et produisent du texte; l'agrégateur reçoit les brouillons comme guidage privé et garde les outils d'origine. L'usage des jetons s'additionne sur tous les appels, et la synthèse continue même si certains proposeurs échouent. + +```mermaid +flowchart TD + R[Requête] --> P1[Proposeur 1] + R --> P2[Proposeur 2] + P1 --> D[Brouillons] + P2 --> D + D --> A[Agrégateur] + A --> O[Réponse finale] +``` + +## Triumvirat + +`createTriumviratModel` reprend l'architecture de Sakana Fugu et de TRINITY. À chaque tour, un coordinateur choisit un modèle dans un vivier interchangeable et lui confie un rôle: le penseur planifie, l'exécutant produit ou corrige la réponse et reçoit seul les outils, le vérificateur juge le brouillon et décide de l'arrêt. La boucle tourne jusqu'à l'acceptation ou jusqu'à une limite de tours. + +Le coordinateur par défaut s'appuie sur un modèle du vivier, avec un repli déterministe penseur, exécutant, vérificateur. Une fonction `decide` fournie par l'appelant remplace ce coordinateur sans changer l'interface. + +## Configurer les ensembles depuis les réglages + +Un bloc `ensembles` dans `.ai/studio.settings.json` rend ces compositions disponibles sans écrire de code. Chaque entrée nomme un `type` (`moa` ou `triumvirat`) et des références de membres au format `/`. Le nom de l'ensemble s'utilise ensuite partout où une référence de modèle s'utilise, par exemple dans `routing.model`. Une configuration incorrecte échoue avec un message clair. + +## Supervision avec Langfuse + +`createLangfuseModel` enveloppe n'importe quel modèle et trace chaque appel vers Langfuse: entrée, sortie, jetons, durée et erreurs. L'enveloppe n'ajoute aucune dépendance: elle écrit vers l'interface publique d'ingestion par `fetch`. L'envoi se fait en arrière-plan et n'ajoute pas de latence à l'appel; `flush()` vide les envois en cours avant la fin d'un processus court. Un échec de supervision n'interrompt jamais l'appel supervisé. Les clés viennent des variables d'environnement `LANGFUSE_PUBLIC_KEY` et `LANGFUSE_SECRET_KEY`, et un hôte auto-hébergé se déclare par `LANGFUSE_HOST`. + +## Choisir entre mixture et triumvirat + +La mixture vise la largeur en un seul tour: des propositions parallèles puis une synthèse, simple et économe. Le triumvirat vise la profondeur en séquence: planifier, produire, vérifier, recommencer, avec autocorrection et usage des outils. Les deux partagent le même port, donc la supervision Langfuse les enveloppe de la même façon. diff --git a/packages/base-llm/README.md b/packages/base-llm/README.md index 5b28e15c..bac636d0 100644 --- a/packages/base-llm/README.md +++ b/packages/base-llm/README.md @@ -24,3 +24,14 @@ console.log(getText(message)); ``` Tool-calling, typed errors (`.code`, `.retriable`), bounded retries with jitter, deadlines and abort are all handled in one tested transport layer (mirrors `@ai-swiss/base-ranker-semantic`). Choosing a model is always an explicit decision — there is no "default/best model" helper. + +## Composition: ensembles and monitoring + +Because every model is the same port, models compose. These meta-models each wrap one or more models and are themselves models, so they drop in anywhere a single model does (settings, Studio, the CLI): + +- [Mixture of Agents](docs/moa.md) — `createMoaModel`: N proposers answer in parallel, one aggregator synthesizes their drafts. +- [Triumvirat](docs/triumvirat.md) — `createTriumviratModel`: a coordinator assigns Thinker / Worker / Verifier roles over a swappable pool, looping until a verifier accepts (the Sakana Fugu / TRINITY architecture). +- [Ensembles in settings](docs/ensembles-settings.md) — configure MoA and Triumvirat from `.ai/studio.settings.json` without code. +- [Langfuse monitoring](docs/langfuse.md) — `createLangfuseModel`: wrap any model to trace input, output, token usage, latency and errors to Langfuse, with zero added dependencies. + +Each page includes a diagram, an options table, and a runnable snippet. diff --git a/packages/base-llm/docs/ensembles-settings.md b/packages/base-llm/docs/ensembles-settings.md new file mode 100644 index 00000000..03169763 --- /dev/null +++ b/packages/base-llm/docs/ensembles-settings.md @@ -0,0 +1,65 @@ +# Configuring ensembles in settings + +MoA (`createMoaModel`) and Triumvirat (`createTriumviratModel`) are usable from +`.ai/studio.settings.json` without code. Add an `ensembles` block; each entry +names a `type` and member refs. A member ref is the same `/` +string a normal model uses, so members are resolved through the existing +provider registry. The ensemble name then works anywhere a model ref does +(for example `routing.model`). + +```mermaid +flowchart LR + REF[model ref] --> Q{name in
ensembles?} + Q -->|no| P[provider/model
single model] + Q -->|yes, moa| M[resolve members
createMoaModel] + Q -->|yes, triumvirat| T[resolve pool
createTriumviratModel] +``` + +## Example + +```json +{ + "providers": [ + { "id": "openai", "type": "openai-compatible", "baseUrl": "https://api.openai.com/v1", "apiKeyEnv": "OPENAI_API_KEY" }, + { "id": "openrouter", "type": "openai-compatible", "baseUrl": "https://openrouter.ai/api/v1", "apiKeyEnv": "OPENROUTER_API_KEY" } + ], + "ensembles": { + "council": { + "type": "moa", + "proposers": ["openai/gpt-5.5", "openrouter/deepseek/deepseek-v4-pro"], + "aggregator": "openrouter/qwen3-max" + }, + "trio": { + "type": "triumvirat", + "pool": { + "deepseek": "openrouter/deepseek/deepseek-v4-pro", + "gpt": "openai/gpt-5.5", + "qwen": "openrouter/qwen3-max" + }, + "maxTurns": 6 + } + } +} +``` + +## Fields + +`type: "moa"` + +| Field | Required | Meaning | +| --- | --- | --- | +| `proposers` | yes | List of member refs that draft in parallel. | +| `aggregator` | yes | Member ref that synthesizes the drafts. | +| `synthesisPrompt` | no | Override the aggregator guidance. | + +`type: "triumvirat"` + +| Field | Required | Meaning | +| --- | --- | --- | +| `pool` | yes | Object of member refs, keyed by name, cheapest-first. | +| `coordinator` | no | Member ref for a dedicated coordinator. Defaults to the cheapest pool member. | +| `maxTurns` | no | Turn budget (default 6). | + +Member refs resolve through the same registry as any model, so a missing +provider, an unknown ensemble type, or a missing required field fails with a +clear `BAD_REQUEST` error. diff --git a/packages/base-llm/docs/langfuse.md b/packages/base-llm/docs/langfuse.md new file mode 100644 index 00000000..3e748d29 --- /dev/null +++ b/packages/base-llm/docs/langfuse.md @@ -0,0 +1,75 @@ +# Langfuse monitoring + +`createLangfuseModel` wraps any base-llm model and conforms to the same port +(`{ id, complete, stream? }`), so it composes over a single adapter, a MoA, or a +Triumvirat. Every call is traced to [Langfuse](https://langfuse.com): input, +output, token usage, latency, and errors. + +Zero new dependencies: no Langfuse SDK, just native `fetch` against the public +ingestion API. Ingestion is fire-and-forget, so it never adds latency to the +model call. A telemetry failure never breaks the wrapped call. + +## Flow + +```mermaid +sequenceDiagram + participant App + participant W as Langfuse wrapper + participant M as Inner model + participant L as Langfuse API + App->>W: complete(request) + W->>M: complete(request) + M-->>W: completion (or error) + W-->>App: completion (or rethrow) + W-)L: POST trace + generation (fire-and-forget) + Note over W,L: flush() drains in-flight sends +``` + +## Usage + +```js +import { createLangfuseModel, createOpenAICompatibleModel } from "@ai-swiss/base-llm"; + +const model = createLangfuseModel({ + model: createOpenAICompatibleModel({ model: "gpt-5.5" }), + // publicKey / secretKey default to env LANGFUSE_PUBLIC_KEY / LANGFUSE_SECRET_KEY +}); + +const out = await model.complete({ messages: [userMessage("Explain CRDTs.")] }); + +// Before a short-lived process exits, drain in-flight telemetry: +await model.flush(); +``` + +Because it wraps any model, you can monitor an ensemble by wrapping it: + +```js +const monitored = createLangfuseModel({ model: createMoaModel({ proposers, aggregator }) }); +``` + +## Options + +| Option | Default | Meaning | +| --- | --- | --- | +| `model` | (required) | The model to wrap. | +| `publicKey` | env `LANGFUSE_PUBLIC_KEY` | Langfuse public key. | +| `secretKey` | env `LANGFUSE_SECRET_KEY` | Langfuse secret key. | +| `baseUrl` | env `LANGFUSE_HOST` or `https://cloud.langfuse.com` | Langfuse host (self-hosted supported). | +| `fetch` | `globalThis.fetch` | Injectable fetch (used in tests). | +| `traceName` | the model id | Name on the emitted trace. | +| `metadata` | none | Static metadata merged into every trace. | +| `onError` | swallow | Called with any ingestion failure. | + +`flush()` resolves once all in-flight ingestion sends complete. `stream()` is +exposed only when the wrapped model streams; it accumulates the streamed text and +the final usage chunk for the generation record. + +## What gets sent + +Per call, one batch with two events to `POST {baseUrl}/api/public/ingestion` +(HTTP Basic auth): + +- `trace-create` with the input messages and the output text. +- `generation-create` with the model id, input, output, mapped token usage + (`{ input, output, total, unit: "TOKENS" }`), `startTime`/`endTime`, and on + failure `level: "ERROR"` plus the error message. diff --git a/packages/base-llm/docs/moa.md b/packages/base-llm/docs/moa.md new file mode 100644 index 00000000..3812fdf3 --- /dev/null +++ b/packages/base-llm/docs/moa.md @@ -0,0 +1,73 @@ +# Mixture of Agents (MoA) + +`createMoaModel` wraps N proposer models and one aggregator into a single model +that conforms to the base-llm port (`{ id, complete, stream? }`). It drops in +anywhere a normal model is used: settings, Studio, the CLI. + +It follows the Mixture-of-Agents pattern (Wang et al. 2024), the shape Sakana +Hermes' `moa` provider uses: many models answer in parallel, then one model +synthesizes their drafts into the final answer. There is no ground-truth judge +in chat, so the aggregator LLM is the selector. + +## Flow + +```mermaid +flowchart TD + R[Request
messages + tools] --> P1[Proposer 1
tools stripped] + R --> P2[Proposer 2
tools stripped] + R --> P3[Proposer N
tools stripped] + P1 --> D[Drafts] + P2 --> D + P3 --> D + D --> A[Aggregator
drafts folded into system message
original tools kept] + A --> O[Final answer
usage summed across all calls] +``` + +Key rules: + +- Proposers run in parallel and receive the request with tools removed. They + draft prose; they do not act. Merging N independent tool-call sets is + undefined, so only the aggregator gets tools. +- The aggregator receives the original conversation plus the drafts as private + guidance folded into the system message (never shown to the user), with the + original tools intact so tool-calling still works. +- If some proposers fail, MoA synthesizes from the survivors. It throws only if + every proposer fails. +- `usage` is summed across all proposer and aggregator calls. + +## Usage + +```js +import { + createMoaModel, + createOpenAICompatibleModel, +} from "@ai-swiss/base-llm"; + +const moa = createMoaModel({ + proposers: [ + createOpenAICompatibleModel({ model: "gpt-5.5" }), + createOpenAICompatibleModel({ baseUrl: openrouter, model: "deepseek/deepseek-v4-pro" }), + ], + aggregator: createOpenAICompatibleModel({ baseUrl: openrouter, model: "qwen3-max" }), +}); + +const out = await moa.complete({ messages: [userMessage("Explain CRDTs.")] }); +``` + +## Options + +| Option | Default | Meaning | +| --- | --- | --- | +| `proposers` | (required) | Non-empty list of models that draft in parallel. | +| `aggregator` | (required) | Model that synthesizes the drafts. | +| `id` | `"moa"` | Value reported as `model.id`. | +| `synthesisPrompt` | `DEFAULT_SYNTHESIS_PROMPT` | Aggregator guidance preamble. | + +`stream()` is exposed only when the aggregator supports streaming; proposers are +gathered first, then the aggregator streams. + +## MoA vs Triumvirat + +MoA is one-shot breadth: parallel proposals, one synthesis pass, cheap. For +sequential plan / execute / verify with self-correction, see Triumvirat +(`createTriumviratModel`). diff --git a/packages/base-llm/docs/triumvirat.md b/packages/base-llm/docs/triumvirat.md new file mode 100644 index 00000000..55620ebb --- /dev/null +++ b/packages/base-llm/docs/triumvirat.md @@ -0,0 +1,99 @@ +# Triumvirat orchestrator + +`createTriumviratModel` implements the inference-time architecture of Sakana Fugu / +TRINITY (arXiv:2512.04695) as a base-llm model. The name nods to the three roles +(Thinker, Worker, Verifier) it coordinates. It conforms to the port +(`{ id, complete }`), so a coordinated pool of models is usable anywhere a single +model is. + +Each turn a coordinator picks one model from a swappable pool and assigns it a +role. The loop runs until the Verifier accepts or a turn budget is hit. + +- Thinker: plans and decomposes. No tools, no final answer. +- Worker: produces or fixes the answer. The only role that receives the + request's tools. +- Verifier: judges the current draft, returns `{ accepted, feedback }`. On + reject, the feedback feeds the next Worker turn. + +## What is and is not reproduced + +TRINITY's contribution is a ~0.6B coordinator trained with separable CMA-ES. +That training is out of scope for a library. The default coordinator here is +prompted (an LLM choosing model + role + stop), with a deterministic +Thinker / Worker / Verifier cadence as the fallback. The `coordinator.decide` +seam lets a heuristic or trained coordinator drop in later with no API change. + +## The turn loop + +```mermaid +flowchart TD + S[Request] --> C{Coordinator
pick model + role} + C -->|thinker| T[Thinker
plan, no tools] + C -->|worker| W[Worker
produce/fix draft
gets tools] + C -->|verifier| V[Verifier
judge draft] + T --> SC[Update shared scratchpad] + W --> SC + V --> J{accepted?} + J -->|yes| DONE[Return draft
finishReason: stop] + J -->|no, feedback| SC + SC --> B{turn < maxTurns?} + B -->|yes| C + B -->|no| INC[Return best draft
finishReason: incomplete] +``` + +A failed role turn is recorded as rejected feedback so the loop can recover +rather than crash. The abort signal is checked between turns. `usage` is summed +across the coordinator and every role call. `raw` carries `turns`, `accepted`, +the last verifier feedback, and the full transcript. + +## Usage + +```js +import { + createTriumviratModel, + createOpenAICompatibleModel, +} from "@ai-swiss/base-llm"; + +const triumvirat = createTriumviratModel({ + // Order cheapest-first: the default coordinator uses the first entry. + pool: { + deepseek: createOpenAICompatibleModel({ baseUrl: openrouter, model: "deepseek/deepseek-v4-pro" }), + gpt: createOpenAICompatibleModel({ model: "gpt-5.5" }), + qwen: createOpenAICompatibleModel({ baseUrl: openrouter, model: "qwen3-max" }), + }, + maxTurns: 6, +}); + +const out = await triumvirat.complete({ messages: [userMessage("Write and verify a binary search.")] }); +``` + +## Options + +| Option | Default | Meaning | +| --- | --- | --- | +| `pool` | (required) | Non-empty map of pool models, keyed by name, cheapest-first. | +| `coordinator` | prompted over cheapest | A dedicated model, or a custom `{ decide(state, ctx) }`. | +| `maxTurns` | `6` | Hard turn budget; guarantees termination. | +| `id` | `"triumvirat"` | Value reported as `model.id`. | +| `rolePrompts` | `DEFAULT_ROLE_PROMPTS` | `{ thinker, worker, verifier }` preambles. | + +## A custom coordinator + +The `decide` seam takes the loop state and returns the next move: + +```js +const heuristic = { + decide(state) { + if (state.turn === 0) return { model: "deepseek", role: "thinker" }; + if (!state.lastDraft) return { model: "gpt", role: "worker" }; + return { model: "qwen", role: "verifier" }; + }, +}; +createTriumviratModel({ pool, coordinator: heuristic }); +``` + +## Triumvirat vs MoA + +Triumvirat is sequential depth: plan, execute, verify, repeat, self-correcting and +tool-using. For one-shot breadth (parallel proposals, one synthesis pass), see +MoA (`createMoaModel`). diff --git a/packages/base-llm/index.mjs b/packages/base-llm/index.mjs index c88160e5..51c407ff 100644 --- a/packages/base-llm/index.mjs +++ b/packages/base-llm/index.mjs @@ -35,4 +35,7 @@ export { createOpenAICompatibleModel, createOllamaModel } from "./src/openai.mjs export { createAnthropicModel } from "./src/anthropic.mjs"; export { createGoogleModel } from "./src/google.mjs"; export { createFauxModel } from "./src/faux.mjs"; +export { createMoaModel, DEFAULT_SYNTHESIS_PROMPT } from "./src/moa.mjs"; +export { createTriumviratModel, DEFAULT_ROLE_PROMPTS } from "./src/triumvirat.mjs"; +export { createLangfuseModel } from "./src/langfuse.mjs"; export { collectStream } from "./src/streaming.mjs"; diff --git a/packages/base-llm/src/langfuse.mjs b/packages/base-llm/src/langfuse.mjs new file mode 100644 index 00000000..72dd7483 --- /dev/null +++ b/packages/base-llm/src/langfuse.mjs @@ -0,0 +1,185 @@ +// Langfuse monitoring wrapper. +// +// Wraps any base-llm model and conforms to the same port ({ id, complete, stream? }), +// so it composes over a single adapter, a MoA, or a Triumvirat. On each call it emits +// a trace + generation to Langfuse's public ingestion API, capturing the input, output, +// token usage, latency, and errors. +// +// Zero new dependencies: no Langfuse SDK, just native fetch against +// `${baseUrl}/api/public/ingestion` with HTTP Basic auth. Ingestion is fire-and-forget +// so it never adds latency to the model call; await `flush()` (e.g. before a CLI exits) +// to drain in-flight sends. Telemetry failures never break the wrapped call. + +import { randomUUID } from "node:crypto"; +import { LlmConfigError } from "./errors.mjs"; +import { getText } from "./types.mjs"; + +const DEFAULT_BASE_URL = "https://cloud.langfuse.com"; + +function iso(ms) { + return new Date(ms).toISOString(); +} + +function mapUsage(usage) { + if (!usage) return undefined; + const input = typeof usage.input === "number" ? usage.input : undefined; + const output = typeof usage.output === "number" ? usage.output : undefined; + if (input === undefined && output === undefined) return undefined; + return { + ...(input !== undefined ? { input } : {}), + ...(output !== undefined ? { output } : {}), + ...(input !== undefined && output !== undefined ? { total: input + output } : {}), + unit: "TOKENS", + }; +} + +/** + * Wrap a model so every call is traced to Langfuse. + * + * @param {object} [options] + * @param {{ id?: string, complete: Function, stream?: Function }} [options.model] Inner model (required). + * @param {string} [options.publicKey] Defaults to env LANGFUSE_PUBLIC_KEY. + * @param {string} [options.secretKey] Defaults to env LANGFUSE_SECRET_KEY. + * @param {string} [options.baseUrl] Langfuse host. Defaults to env LANGFUSE_HOST or cloud.langfuse.com. + * @param {*} [options.fetch] fetch implementation (default globalThis.fetch). + * @param {*} [options.env] Environment object (default process.env). + * @param {string} [options.traceName] Trace name (default the model id). + * @param {object} [options.metadata] Static metadata merged into every trace. + * @param {(error: unknown) => void} [options.onError] Called on ingestion failure (default: swallow). + * @returns {{ id: string, complete: Function, stream?: Function, flush: () => Promise }} + */ +export function createLangfuseModel(options = {}) { + const { + model, + baseUrl, + fetch: fetchImpl, + env = process.env, + traceName, + metadata, + onError, + } = options; + + if (!model || typeof model.complete !== "function") { + throw new LlmConfigError("createLangfuseModel requires a `model` with complete()"); + } + // Bind the validated model so it reads as defined inside the closures below. + const innerModel = model; + const publicKey = options.publicKey ?? env.LANGFUSE_PUBLIC_KEY; + const secretKey = options.secretKey ?? env.LANGFUSE_SECRET_KEY; + if (!publicKey || !secretKey) { + throw new LlmConfigError( + "createLangfuseModel requires publicKey/secretKey (or env LANGFUSE_PUBLIC_KEY / LANGFUSE_SECRET_KEY)", + ); + } + const host = baseUrl ?? env.LANGFUSE_HOST ?? DEFAULT_BASE_URL; + const doFetch = fetchImpl ?? globalThis.fetch; + if (typeof doFetch !== "function") { + throw new LlmConfigError("createLangfuseModel: no fetch available; pass options.fetch"); + } + const auth = `Basic ${btoa(`${publicKey}:${secretKey}`)}`; + const name = traceName ?? innerModel.id ?? "base-llm"; + + /** @type {Set>} in-flight ingestion sends, drained by flush(). */ + const pending = new Set(); + + /** @param {{ request: any, output?: any, usage?: any, start: number, end: number, error?: any }} args */ + function send({ request, output, usage, start, end, error }) { + const traceId = randomUUID(); + const events = [ + { + id: randomUUID(), + type: "trace-create", + timestamp: iso(start), + body: { + id: traceId, + name, + timestamp: iso(start), + input: request.messages, + ...(output !== undefined ? { output } : {}), + ...(metadata ? { metadata } : {}), + }, + }, + { + id: randomUUID(), + type: "generation-create", + timestamp: iso(end), + body: { + id: randomUUID(), + traceId, + name: "completion", + model: innerModel.id, + input: request.messages, + ...(output !== undefined ? { output } : {}), + ...(mapUsage(usage) ? { usage: mapUsage(usage) } : {}), + startTime: iso(start), + endTime: iso(end), + ...(error ? { level: "ERROR", statusMessage: String(error?.message ?? error) } : {}), + }, + }, + ]; + + const promise = (async () => { + try { + const res = await doFetch(`${host}/api/public/ingestion`, { + method: "POST", + headers: { "Content-Type": "application/json", Authorization: auth }, + body: JSON.stringify({ batch: events }), + }); + if (res && res.ok === false) throw new Error(`langfuse ingestion failed: HTTP ${res.status}`); + } catch (err) { + if (onError) onError(err); + } + })(); + pending.add(promise); + promise.finally(() => pending.delete(promise)); + } + + async function complete(request, ctx = {}) { + const start = Date.now(); + let completion; + try { + completion = await innerModel.complete(request, ctx); + } catch (error) { + send({ request, output: undefined, usage: undefined, start, end: Date.now(), error }); + throw error; + } + send({ + request, + output: getText(completion.message), + usage: completion.usage, + start, + end: Date.now(), + }); + return completion; + } + + async function* stream(request, ctx = {}) { + const innerStream = innerModel.stream; + if (typeof innerStream !== "function") { + throw new LlmConfigError("createLangfuseModel: wrapped model does not support streaming"); + } + const start = Date.now(); + let text = ""; + let usage; + try { + for await (const chunk of innerStream.call(innerModel, request, ctx)) { + if (chunk && typeof chunk.text === "string") text += chunk.text; + if (chunk && chunk.usage) usage = chunk.usage; + yield chunk; + } + } catch (error) { + send({ request, output: text || undefined, usage, start, end: Date.now(), error }); + throw error; + } + send({ request, output: text, usage, start, end: Date.now() }); + } + + async function flush() { + await Promise.all([...pending]); + } + + /** @type {{ id: string, complete: Function, stream?: Function, flush: () => Promise }} */ + const wrapped = { id: innerModel.id ?? "langfuse", complete, flush }; + if (typeof innerModel.stream === "function") wrapped.stream = stream; + return wrapped; +} diff --git a/packages/base-llm/src/moa.mjs b/packages/base-llm/src/moa.mjs new file mode 100644 index 00000000..a290e6cc --- /dev/null +++ b/packages/base-llm/src/moa.mjs @@ -0,0 +1,139 @@ +// Mixture-of-Agents (MoA) meta-model. +// +// Wraps N "proposer" models plus one "aggregator" model and conforms to the same +// LanguageModel port as every other adapter ({ id, complete, stream? }), so it +// drops in anywhere a single model is expected — settings, Studio, the CLI. +// +// Flow (Wang et al. 2024 "Mixture-of-Agents", the shape Hermes' `moa` provider uses): +// 1. Every proposer answers the SAME request in parallel. Proposers see the request +// with tools stripped — they produce text drafts, they do not act. This keeps +// merging well-defined (you can synthesize prose; you cannot meaningfully merge +// N independent tool-call sets) and keeps proposers cheap. +// 2. The aggregator receives the original conversation plus the drafts as private +// guidance (folded into the system message, never shown to the user) and the +// ORIGINAL tools intact, so tool-calling still works through the aggregator. +// 3. The aggregator's response is the result. Usage is summed across all models. +// +// There is no ground-truth judge here (unlike a compiler/benchmark loop), so the +// aggregator LLM is the selector — exactly Hermes' design for chat assistants. + +import { LlmConfigError, LlmResponseError } from "./errors.mjs"; +import { assertValidRequest, getText, systemMessage } from "./types.mjs"; + +export const DEFAULT_SYNTHESIS_PROMPT = + "You have been given candidate responses from several assistant models for the " + + "latest user request, labelled below. Synthesize them into a single, higher-quality " + + "answer. Do not copy one verbatim: weigh them critically, correct mistakes, and keep " + + "the strongest reasoning from each. Treat the candidates as advice, not fact — if they " + + "conflict, decide which is right. Respond directly to the user and never mention that " + + "multiple candidates existed."; + +/** + * @param {unknown} m + * @returns {m is {complete: Function, stream?: Function}} + */ +function isModel(m) { + return Boolean(m) && typeof (/** @type {any} */ (m)).complete === "function"; +} + +// Sum any numeric usage fields without assuming a fixed shape ({ input, output } +// today, but defensive against adapters that report extra counters). +function addUsage(acc, usage) { + if (!usage) return acc; + for (const [key, value] of Object.entries(usage)) { + if (typeof value === "number") acc[key] = (acc[key] ?? 0) + value; + } + return acc; +} + +/** + * Create a Mixture-of-Agents model. + * + * @param {object} [options] + * @param {Array<{complete: Function, stream?: Function}>} [options.proposers] + * Non-empty list of models that draft candidate responses in parallel. + * @param {{complete: Function, stream?: Function}} [options.aggregator] + * Model that synthesizes the drafts into the final response. + * @param {string} [options.id="moa"] Identifier reported as `model.id`. + * @param {string} [options.synthesisPrompt] Override the aggregator guidance. + * @returns {{id: string, complete: Function, stream?: Function}} + */ +export function createMoaModel(options = {}) { + const { + proposers, + aggregator, + id = "moa", + synthesisPrompt = DEFAULT_SYNTHESIS_PROMPT, + } = options; + + if (!Array.isArray(proposers) || proposers.length === 0) { + throw new LlmConfigError("createMoaModel requires a non-empty `proposers` array"); + } + proposers.forEach((p, i) => { + if (!isModel(p)) throw new LlmConfigError(`createMoaModel proposers[${i}] is not a model (missing complete())`); + }); + if (!isModel(aggregator)) { + throw new LlmConfigError("createMoaModel requires an `aggregator` model with complete()"); + } + // Bind the validated values so they read as defined inside the closures below. + const panel = proposers; + const merge = aggregator; + + async function gatherDrafts(request, ctx) { + // Proposers draft prose only — strip tools so nobody tries to act. + const proposerRequest = { ...request, tools: undefined }; + const settled = await Promise.all( + panel.map((p) => + p.complete(proposerRequest, ctx).then( + (c) => ({ ok: true, text: (getText(c.message) ?? "").trim(), usage: c.usage }), + (error) => ({ ok: false, error }), + ), + ), + ); + const drafts = settled.filter((s) => s.ok && s.text); + if (drafts.length === 0) { + const firstError = settled.find((s) => !s.ok)?.error; + throw firstError ?? new LlmResponseError("moa: every proposer returned an empty response"); + } + return { drafts, settled }; + } + + function aggregatorRequest(request, drafts) { + const guidance = + synthesisPrompt + + "\n\n" + + drafts.map((d, i) => `### Candidate response ${i + 1}\n${d.text}`).join("\n\n"); + const [first, ...rest] = request.messages; + const messages = + first && first.role === "system" + ? [systemMessage(`${getText(first)}\n\n${guidance}`), ...rest] + : [systemMessage(guidance), ...request.messages]; + // Aggregator keeps the ORIGINAL tools so tool-calling survives. + return { ...request, messages }; + } + + async function complete(request, ctx = {}) { + assertValidRequest(request); + const { drafts, settled } = await gatherDrafts(request, ctx); + const final = await merge.complete(aggregatorRequest(request, drafts), ctx); + const usage = addUsage({}, final.usage); + for (const s of settled) if (s.ok) addUsage(usage, s.usage); + return { ...final, usage }; + } + + async function* stream(request, ctx = {}) { + const aggStream = merge.stream; + if (typeof aggStream !== "function") { + throw new LlmConfigError("moa: aggregator does not support streaming"); + } + assertValidRequest(request); + const { drafts } = await gatherDrafts(request, ctx); + // Only the aggregator streams; proposers are gathered up front. + yield* aggStream.call(merge, aggregatorRequest(request, drafts), ctx); + } + + /** @type {{id: string, complete: Function, stream?: Function}} */ + const model = { id, complete }; + if (typeof merge.stream === "function") model.stream = stream; + return model; +} diff --git a/packages/base-llm/src/triumvirat.mjs b/packages/base-llm/src/triumvirat.mjs new file mode 100644 index 00000000..ce4f2aa7 --- /dev/null +++ b/packages/base-llm/src/triumvirat.mjs @@ -0,0 +1,256 @@ +// Triumvirat orchestrator meta-model (implements the TRINITY architecture). +// +// Implements the inference-time architecture of Sakana Fugu / TRINITY +// (arXiv:2512.04695) as a base-llm meta-model conforming to the LanguageModel +// port ({ id, complete }), so a coordinated pool of models is usable anywhere a +// single model is. +// +// Each turn a coordinator picks one model from a swappable pool and assigns it +// one of three roles, looping until the Verifier accepts or a turn budget hits: +// - Thinker — plans/decomposes (no tools, no final answer) +// - Worker — executes; the ONLY role that receives the request's tools +// - Verifier — judges the current draft, returns { accepted, feedback } +// +// What this is NOT: TRINITY's contribution is a ~0.6B coordinator trained with +// separable CMA-ES. We can't reproduce that training, so the default coordinator +// is *prompted* (an LLM asked to choose model+role+stop), with a deterministic +// Thinker -> Worker -> Verifier cadence as the fallback. The `coordinator.decide` +// seam lets a heuristic or trained coordinator drop in later with no API change. + +import { LlmConfigError, LlmResponseError, LlmAbortError } from "./errors.mjs"; +import { assertValidRequest, getText, systemMessage } from "./types.mjs"; +import { extractJson } from "./json.mjs"; + +const ROLES = new Set(["thinker", "worker", "verifier"]); + +export const DEFAULT_ROLE_PROMPTS = { + thinker: + "You are the THINKER. Decompose the task and outline an approach. Do not write " + + "the final answer and do not call tools. Keep it short and concrete.", + worker: + "You are the WORKER. Produce or improve the answer to the user's request, using " + + "the shared scratchpad and any verifier feedback. If tools are available, use them. " + + "Output the answer itself, nothing else.", + verifier: + "You are the VERIFIER. Judge whether the latest draft fully and correctly answers " + + "the user's request. Respond ONLY with JSON: " + + '{"accepted": boolean, "feedback": string}. ' + + "Set accepted=true only if it is complete and correct; otherwise give specific, " + + "actionable feedback for the next worker turn.", +}; + +function isModel(m) { + return Boolean(m) && typeof (/** @type {any} */ (m)).complete === "function"; +} + +// Sum numeric usage fields without assuming a fixed shape. +// (Duplicated from moa.mjs on purpose to keep this PR self-contained; a shared +// internal/ensemble.mjs is a follow-up once both have landed.) +function addUsage(acc, usage) { + if (!usage) return acc; + for (const [key, value] of Object.entries(usage)) { + if (typeof value === "number") acc[key] = (acc[key] ?? 0) + value; + } + return acc; +} + +function withSystem(preamble, messages) { + const [first, ...rest] = messages; + return first && first.role === "system" + ? [systemMessage(`${getText(first)}\n\n${preamble}`), ...rest] + : [systemMessage(preamble), ...messages]; +} + +/** + * @param {{transcript: Array<{turn:number,model:string,role:string,text:string}>, verdict: any}} state + */ +function buildScratchpad(state) { + if (state.transcript.length === 0 && !state.verdict) return "Shared scratchpad: (empty)"; + const lines = state.transcript.map( + (t) => `- turn ${t.turn} [${t.role} via ${t.model}]: ${t.text}`, + ); + if (state.verdict && !state.verdict.accepted && state.verdict.feedback) { + lines.push(`- verifier feedback to address: ${state.verdict.feedback}`); + } + return `Shared scratchpad so far:\n${lines.join("\n")}`; +} + +function coordinatorPrompt(poolKeys, state) { + return ( + "You are the COORDINATOR of a pool of models. Each turn, choose which model " + + "answers next and in which role.\n" + + `Pool (choose one key): ${poolKeys.join(", ")}\n` + + "Roles: thinker (plan), worker (produce/fix the answer), verifier (judge the draft).\n" + + "Pick worker before verifier; use verifier to gate completion.\n" + + 'Respond ONLY with JSON: {"model": , "role": , "stop": }.\n\n' + + `Turn: ${state.turn}. Draft exists: ${Boolean(state.lastDraft)}. ` + + `Last verdict: ${state.verdict ? JSON.stringify(state.verdict) : "none"}.\n` + + buildScratchpad(state) + ); +} + +/** + * Create a Triumvirat orchestrator model (implements the TRINITY architecture). + * + * @param {object} [options] + * @param {Record} [options.pool] + * Non-empty map of pool models, keyed by name. Order cheapest-first: the + * default coordinator uses the first entry. + * @param {{complete: Function} | {decide: Function}} [options.coordinator] + * A dedicated model (prompted coordinator) or a custom `{ decide(state, ctx) }`. + * Defaults to a prompted coordinator over the cheapest (first) pool model. + * @param {number} [options.maxTurns=6] Hard turn budget; guarantees termination. + * @param {string} [options.id="triumvirat"] + * @param {{thinker:string, worker:string, verifier:string}} [options.rolePrompts] + * @returns {{id: string, complete: Function}} + */ +export function createTriumviratModel(options = {}) { + const { + pool, + coordinator, + maxTurns = 6, + id = "triumvirat", + rolePrompts = DEFAULT_ROLE_PROMPTS, + } = options; + + if (!pool || typeof pool !== "object" || Array.isArray(pool)) { + throw new LlmConfigError("createTriumviratModel requires a `pool` object of models"); + } + // Bind the validated pool so it reads as defined inside the closures below. + const poolMap = pool; + const poolKeys = Object.keys(poolMap); + if (poolKeys.length === 0) { + throw new LlmConfigError("createTriumviratModel requires a non-empty `pool`"); + } + for (const key of poolKeys) { + if (!isModel(poolMap[key])) throw new LlmConfigError(`createTriumviratModel pool["${key}"] is not a model`); + } + if (!Number.isInteger(maxTurns) || maxTurns < 1) { + throw new LlmConfigError("createTriumviratModel `maxTurns` must be an integer >= 1"); + } + + // ponytail: "cheapest" = first pool entry. Order the pool cheapest-first, or + // pass an explicit coordinator model, to control which model coordinates. + /** @type {{complete: Function}} */ + const coordinatorModel = isModel(coordinator) + ? /** @type {{complete: Function}} */ (coordinator) + : /** @type {{complete: Function}} */ (poolMap[poolKeys[0]]); + + // Deterministic fallback policy: thinker on turn 0, then alternate worker/verifier. + function deterministic(state) { + if (state.turn === 0) return { model: poolKeys[0], role: "thinker" }; + return state.turn % 2 === 1 + ? { model: poolKeys[0], role: "worker" } + : { model: poolKeys[0], role: "verifier" }; + } + + /** @type {{decide: Function}} */ + const decider = + coordinator && typeof (/** @type {any} */ (coordinator)).decide === "function" + ? /** @type {{decide: Function}} */ (coordinator) + : { + async decide(state, ctx) { + const messages = withSystem(coordinatorPrompt(poolKeys, state), state.request.messages); + let resp; + try { + resp = await coordinatorModel.complete({ messages, temperature: 0 }, ctx); + } catch { + return deterministic(state); // coordinator failed: keep the loop moving + } + const d = extractJson(getText(resp.message)); + if (d && poolMap[d.model] && ROLES.has(d.role)) { + return { model: d.model, role: d.role, stop: d.stop === true, usage: resp.usage }; + } + return { ...deterministic(state), usage: resp.usage }; + }, + }; + + function roleRequest(role, state) { + const preamble = `${rolePrompts[role]}\n\n${buildScratchpad(state)}`; + const messages = withSystem(preamble, state.request.messages); + const req = { ...state.request, messages, tools: role === "worker" ? state.request.tools : undefined }; + if (role !== "worker") req.temperature = 0; + return req; + } + + async function complete(request, ctx = {}) { + assertValidRequest(request); + /** @type {Record} */ + const usage = {}; + /** + * @type {{ + * request: any, + * transcript: Array<{turn:number, model:string, role:string, text:string}>, + * lastDraft: any, + * verdict: {accepted:boolean, feedback:string} | null, + * turn: number, + * }} + */ + const state = { request, transcript: [], lastDraft: null, verdict: null, turn: 0 }; + + while (state.turn < maxTurns) { + if (ctx.signal?.aborted) throw new LlmAbortError("triumvirat: aborted"); + + const decision = await decider.decide(state, ctx); + if (decision.usage) addUsage(usage, decision.usage); + const model = poolMap[decision.model] ?? /** @type {{complete: Function}} */ (poolMap[poolKeys[0]]); + const role = ROLES.has(decision.role) ? decision.role : "worker"; + + let resp; + try { + resp = await model.complete(roleRequest(role, state), ctx); + } catch (error) { + if (error instanceof LlmAbortError) throw error; + const message = error instanceof Error ? error.message : String(error); + state.transcript.push({ turn: state.turn, model: decision.model, role, text: `ERROR: ${message}` }); + state.verdict = { accepted: false, feedback: `Previous ${role} turn failed: ${message}` }; + state.turn += 1; + continue; + } + addUsage(usage, resp.usage); + const text = (getText(resp.message) ?? "").trim(); + + if (role === "verifier") { + const obj = extractJson(text); + state.verdict = { + accepted: obj?.accepted === true, + feedback: typeof obj?.feedback === "string" ? obj.feedback : "", + }; + state.transcript.push({ + turn: state.turn, + model: decision.model, + role, + text: `accepted=${state.verdict.accepted}${state.verdict.feedback ? ` (${state.verdict.feedback})` : ""}`, + }); + if (state.verdict.accepted) { + state.turn += 1; + break; + } + } else { + state.transcript.push({ turn: state.turn, model: decision.model, role, text }); + if (role === "worker") state.lastDraft = resp.message; + } + + state.turn += 1; + if (decision.stop && state.lastDraft) break; + } + + if (!state.lastDraft) { + throw new LlmResponseError("triumvirat: no answer produced within the turn budget"); + } + const accepted = state.verdict?.accepted === true; + return { + message: state.lastDraft, + usage, + finishReason: accepted ? "stop" : "incomplete", + raw: { + turns: state.turn, + accepted, + verdictFeedback: state.verdict?.feedback ?? null, + transcript: state.transcript, + }, + }; + } + + return { id, complete }; +} diff --git a/packages/base-llm/tests/base-llm.langfuse.test.mjs b/packages/base-llm/tests/base-llm.langfuse.test.mjs new file mode 100644 index 00000000..db2d2321 --- /dev/null +++ b/packages/base-llm/tests/base-llm.langfuse.test.mjs @@ -0,0 +1,132 @@ +// Spec coverage: UR-CORE-003 +import { describe, it } from "node:test"; +import assert from "node:assert/strict"; + +import { + createLangfuseModel, + assistantMessage, + userMessage, + getText, + LlmConfigError, +} from "../index.mjs"; + +// Capturing fake fetch: records each call and returns a canned response. +function fakeFetch(respond = () => ({ ok: true, status: 207 })) { + const calls = []; + const fn = async (url, opts) => { + calls.push({ url, opts, body: JSON.parse(opts.body) }); + return respond(url, opts); + }; + fn.calls = calls; + return fn; +} + +function inner(text, usage = { input: 5, output: 7 }) { + return { + id: "gpt-x", + async complete() { + return { message: assistantMessage(text), usage, finishReason: "stop" }; + }, + }; +} + +const KEYS = { publicKey: "pk", secretKey: "sk" }; +const ask = { messages: [userMessage("hi")] }; + +describe("createLangfuseModel", () => { + it("returns the inner completion unchanged and emits a trace + generation", async () => { + const f = fakeFetch(); + const m = createLangfuseModel({ model: inner("hello"), ...KEYS, fetch: f }); + + const out = await m.complete(ask); + assert.equal(getText(out.message), "hello"); + + await m.flush(); + assert.equal(f.calls.length, 1); + const call = f.calls[0]; + assert.match(call.url, /\/api\/public\/ingestion$/); + assert.equal(call.opts.method, "POST"); + assert.match(call.opts.headers.Authorization, /^Basic /); + assert.deepEqual(call.body.batch.map((e) => e.type), ["trace-create", "generation-create"]); + + const gen = call.body.batch[1].body; + assert.equal(gen.model, "gpt-x"); + assert.equal(gen.output, "hello"); + assert.deepEqual(gen.usage, { input: 5, output: 7, total: 12, unit: "TOKENS" }); + assert.ok(gen.startTime && gen.endTime); + }); + + it("emits an ERROR generation and rethrows when the inner model fails", async () => { + const f = fakeFetch(); + const failing = { id: "x", async complete() { throw new Error("upstream 500"); } }; + const m = createLangfuseModel({ model: failing, ...KEYS, fetch: f }); + + await assert.rejects(() => m.complete(ask), /upstream 500/); + await m.flush(); + assert.equal(f.calls[0].body.batch[1].body.level, "ERROR"); + }); + + it("never breaks the call when ingestion fails (errors routed to onError)", async () => { + const errs = []; + const f = fakeFetch(() => ({ ok: false, status: 401 })); + const m = createLangfuseModel({ model: inner("ok"), ...KEYS, fetch: f, onError: (e) => errs.push(e) }); + + const out = await m.complete(ask); + assert.equal(getText(out.message), "ok"); + await m.flush(); + assert.equal(errs.length, 1); + }); + + it("reads keys from the environment", async () => { + const f = fakeFetch(); + const m = createLangfuseModel({ + model: inner("x"), + fetch: f, + env: { LANGFUSE_PUBLIC_KEY: "pk", LANGFUSE_SECRET_KEY: "sk" }, + }); + await m.complete(ask); + await m.flush(); + assert.match(f.calls[0].opts.headers.Authorization, /^Basic /); + }); + + it("captures streamed text and usage, passing chunks through", async () => { + const f = fakeFetch(); + const streamer = { + id: "s", + async complete() { return { message: assistantMessage("full"), usage: { input: 1, output: 1 } }; }, + async *stream() { + yield { type: "text-delta", text: "he" }; + yield { type: "text-delta", text: "llo" }; + yield { type: "done", usage: { input: 2, output: 3 } }; + }, + }; + const m = createLangfuseModel({ model: streamer, ...KEYS, fetch: f }); + + const chunks = []; + for await (const c of m.stream(ask)) chunks.push(c); + assert.equal(chunks.length, 3); + + await m.flush(); + const gen = f.calls[0].body.batch[1].body; + assert.equal(gen.output, "hello"); + assert.deepEqual(gen.usage, { input: 2, output: 3, total: 5, unit: "TOKENS" }); + }); + + it("exposes stream() only when the wrapped model streams", () => { + const withStream = createLangfuseModel({ + model: { id: "s", complete: inner("x").complete, async *stream() {} }, + ...KEYS, + fetch: fakeFetch(), + }); + assert.equal(typeof withStream.stream, "function"); + + const noStream = createLangfuseModel({ model: inner("x"), ...KEYS, fetch: fakeFetch() }); + assert.equal(typeof noStream.stream, "undefined"); + }); + + it("validates configuration", () => { + assert.throws(() => createLangfuseModel({ ...KEYS }), LlmConfigError); // no model + assert.throws(() => createLangfuseModel({ model: inner("x"), secretKey: "sk", env: {} }), LlmConfigError); // no public key + assert.throws(() => createLangfuseModel({ model: inner("x"), publicKey: "pk", env: {} }), LlmConfigError); // no secret key + }); +}); diff --git a/packages/base-llm/tests/base-llm.moa.test.mjs b/packages/base-llm/tests/base-llm.moa.test.mjs new file mode 100644 index 00000000..53e47954 --- /dev/null +++ b/packages/base-llm/tests/base-llm.moa.test.mjs @@ -0,0 +1,128 @@ +// Spec coverage: UR-CORE-003 +import { describe, it } from "node:test"; +import assert from "node:assert/strict"; + +import { + createMoaModel, + assistantMessage, + userMessage, + getText, + LlmConfigError, +} from "../index.mjs"; + +// Minimal capturing model stub — records the request it received so we can assert +// what the MoA layer forwarded to proposers vs the aggregator. +function stub(text, { usage = { input: 1, output: 1 }, capture, fail } = {}) { + return { + id: "stub", + async complete(request, ctx) { + capture?.(request, ctx); + if (fail) throw fail; + return { message: assistantMessage(text), usage, finishReason: "stop" }; + }, + }; +} + +const baseRequest = () => ({ messages: [userMessage("What is 2+2?")] }); + +describe("createMoaModel", () => { + it("synthesizes proposer drafts via the aggregator and returns the aggregator's answer", async () => { + let aggReq; + const moa = createMoaModel({ + proposers: [stub("draft A says four"), stub("draft B says 4")], + aggregator: stub("The answer is 4.", { capture: (r) => (aggReq = r) }), + }); + + const out = await moa.complete(baseRequest()); + + assert.equal(getText(out.message), "The answer is 4."); + // Aggregator's first message is a system message carrying BOTH drafts. + assert.equal(aggReq.messages[0].role, "system"); + const sys = getText(aggReq.messages[0]); + assert.match(sys, /draft A says four/); + assert.match(sys, /draft B says 4/); + // The original user turn is preserved after the synthesis guidance. + assert.equal(getText(aggReq.messages.at(-1)), "What is 2+2?"); + }); + + it("strips tools from proposers but preserves them for the aggregator", async () => { + const seen = { proposer: "unset", aggregator: "unset" }; + const tools = [{ name: "calc", parameters: { type: "object" } }]; + const moa = createMoaModel({ + proposers: [stub("p", { capture: (r) => (seen.proposer = r.tools) })], + aggregator: stub("a", { capture: (r) => (seen.aggregator = r.tools) }), + }); + + await moa.complete({ ...baseRequest(), tools }); + + assert.equal(seen.proposer, undefined, "proposer must not receive tools"); + assert.deepEqual(seen.aggregator, tools, "aggregator must keep original tools"); + }); + + it("still synthesizes when some proposers fail, dropping the failures", async () => { + let aggReq; + const moa = createMoaModel({ + proposers: [ + stub("", { fail: new Error("boom") }), + stub("the survivor draft"), + ], + aggregator: stub("final", { capture: (r) => (aggReq = r) }), + }); + + const out = await moa.complete(baseRequest()); + + assert.equal(getText(out.message), "final"); + assert.match(getText(aggReq.messages[0]), /the survivor draft/); + }); + + it("throws when every proposer fails", async () => { + const moa = createMoaModel({ + proposers: [stub("", { fail: new Error("a") }), stub("", { fail: new Error("b") })], + aggregator: stub("never reached"), + }); + await assert.rejects(() => moa.complete(baseRequest()), /a/); + }); + + it("sums usage across all proposers and the aggregator", async () => { + const moa = createMoaModel({ + proposers: [ + stub("p1", { usage: { input: 10, output: 5 } }), + stub("p2", { usage: { input: 20, output: 7 } }), + ], + aggregator: stub("agg", { usage: { input: 3, output: 9 } }), + }); + + const out = await moa.complete(baseRequest()); + + assert.deepEqual(out.usage, { input: 33, output: 21 }); + }); + + it("validates configuration", () => { + assert.throws(() => createMoaModel({ proposers: [], aggregator: stub("x") }), LlmConfigError); + assert.throws(() => createMoaModel({ proposers: [stub("x")] }), LlmConfigError); + assert.throws( + () => createMoaModel({ proposers: [{ notAModel: true }], aggregator: stub("x") }), + LlmConfigError, + ); + }); + + it("exposes stream() only when the aggregator can stream, delegating to it", async () => { + const noStream = createMoaModel({ proposers: [stub("p")], aggregator: stub("a") }); + assert.equal(typeof noStream.stream, "undefined"); + + const streamingAggregator = { + id: "agg", + complete: stub("a").complete, + async *stream() { + yield { type: "text-delta", text: "hel" }; + yield { type: "text-delta", text: "lo" }; + }, + }; + const moa = createMoaModel({ proposers: [stub("draft")], aggregator: streamingAggregator }); + assert.equal(typeof moa.stream, "function"); + + const chunks = []; + for await (const c of moa.stream(baseRequest())) chunks.push(c.text); + assert.deepEqual(chunks, ["hel", "lo"]); + }); +}); diff --git a/packages/base-llm/tests/base-llm.triumvirat.test.mjs b/packages/base-llm/tests/base-llm.triumvirat.test.mjs new file mode 100644 index 00000000..7a20d91e --- /dev/null +++ b/packages/base-llm/tests/base-llm.triumvirat.test.mjs @@ -0,0 +1,230 @@ +// Spec coverage: UR-CORE-003 +import { describe, it } from "node:test"; +import assert from "node:assert/strict"; + +import { + createTriumviratModel, + assistantMessage, + userMessage, + getText, + LlmConfigError, +} from "../index.mjs"; + +// --- model stubs (no network) ----------------------------------------------- + +function textModel(text, usage = { input: 1, output: 1 }, capture) { + return { + id: "stub", + async complete(request) { + capture?.(request); + return { message: assistantMessage(text), usage, finishReason: "stop" }; + }, + }; +} + +// Returns the next canned text per call (clamps on the last entry). +function queueModel(texts, usage = { input: 1, output: 1 }) { + let i = 0; + return { + id: "stub", + async complete() { + const text = texts[Math.min(i, texts.length - 1)]; + i += 1; + return { message: assistantMessage(text), usage, finishReason: "stop" }; + }, + }; +} + +function failModel(error) { + return { id: "stub", async complete() { throw error; } }; +} + +// A coordinator that replays a fixed script of decisions (clamps on the last). +function scripted(steps) { + let i = 0; + return { decide() { const s = steps[Math.min(i, steps.length - 1)]; i += 1; return s; } }; +} + +const ask = (extra = {}) => ({ messages: [userMessage("solve it")], ...extra }); + +describe("createTriumviratModel", () => { + it("runs thinker -> worker -> verifier(accept) and returns the worker draft", async () => { + const pool = { + t: textModel("plan: do the thing"), + w: textModel("the answer is 42"), + v: textModel('{"accepted":true,"feedback":""}'), + }; + const triumvirat = createTriumviratModel({ + pool, + coordinator: scripted([ + { model: "t", role: "thinker" }, + { model: "w", role: "worker" }, + { model: "v", role: "verifier" }, + ]), + }); + + const out = await triumvirat.complete(ask()); + + assert.equal(getText(out.message), "the answer is 42"); + assert.equal(out.finishReason, "stop"); + assert.equal(out.raw.accepted, true); + }); + + it("feeds verifier feedback into the next worker turn (reject then fix)", async () => { + let workerCalls = 0; + const workerSystems = []; + const pool = { + w: { + id: "w", + async complete(request) { + workerCalls += 1; + workerSystems.push(getText(request.messages[0])); + return { message: assistantMessage(`draft ${workerCalls}`), usage: { input: 1, output: 1 } }; + }, + }, + v: queueModel(['{"accepted":false,"feedback":"handle the empty case"}', '{"accepted":true}']), + }; + const triumvirat = createTriumviratModel({ + pool, + coordinator: scripted([ + { model: "w", role: "worker" }, + { model: "v", role: "verifier" }, + { model: "w", role: "worker" }, + { model: "v", role: "verifier" }, + ]), + }); + + const out = await triumvirat.complete(ask()); + + assert.equal(out.raw.accepted, true); + assert.equal(workerCalls, 2); + assert.equal(getText(out.message), "draft 2"); + assert.match(workerSystems[1], /handle the empty case/); + }); + + it("stops at the turn budget and returns the best draft when never accepted", async () => { + const pool = { + w: textModel("partial answer"), + v: textModel('{"accepted":false,"feedback":"not yet"}'), + }; + const triumvirat = createTriumviratModel({ + pool, + coordinator: scripted([{ model: "w", role: "worker" }, { model: "v", role: "verifier" }]), + maxTurns: 4, + }); + + const out = await triumvirat.complete(ask()); + + assert.equal(out.finishReason, "incomplete"); + assert.equal(out.raw.accepted, false); + assert.equal(getText(out.message), "partial answer"); + assert.ok(out.raw.turns <= 4); + }); + + it("gives tools only to the worker role", async () => { + const seen = {}; + const pool = { + t: textModel("plan", undefined, (r) => (seen.thinker = r.tools)), + w: textModel("answer", undefined, (r) => (seen.worker = r.tools)), + v: textModel('{"accepted":true}', undefined, (r) => (seen.verifier = r.tools)), + }; + const tools = [{ name: "calc", parameters: { type: "object" } }]; + const triumvirat = createTriumviratModel({ + pool, + coordinator: scripted([ + { model: "t", role: "thinker" }, + { model: "w", role: "worker" }, + { model: "v", role: "verifier" }, + ]), + }); + + await triumvirat.complete(ask({ tools })); + + assert.equal(seen.thinker, undefined); + assert.deepEqual(seen.worker, tools); + assert.equal(seen.verifier, undefined); + }); + + it("sums usage across every role call", async () => { + const pool = { + w: textModel("a", { input: 10, output: 5 }), + v: textModel('{"accepted":true}', { input: 2, output: 3 }), + }; + const triumvirat = createTriumviratModel({ + pool, + coordinator: scripted([{ model: "w", role: "worker" }, { model: "v", role: "verifier" }]), + }); + + const out = await triumvirat.complete(ask()); + + assert.deepEqual(out.usage, { input: 12, output: 8 }); + }); + + it("drives to acceptance through a prompted (LLM) coordinator", async () => { + const coordinator = queueModel(['{"model":"w","role":"worker"}', '{"model":"v","role":"verifier"}']); + const pool = { w: textModel("done"), v: textModel('{"accepted":true}') }; + const triumvirat = createTriumviratModel({ pool, coordinator, maxTurns: 5 }); + + const out = await triumvirat.complete(ask()); + + assert.equal(out.raw.accepted, true); + assert.equal(getText(out.message), "done"); + }); + + it("falls back to the deterministic policy when the coordinator emits junk", async () => { + // Coordinator returns non-JSON every turn -> deterministic thinker/worker/verifier. + const pool = { a: textModel("the answer") }; + const triumvirat = createTriumviratModel({ pool, coordinator: textModel("not json"), maxTurns: 3 }); + + const out = await triumvirat.complete(ask()); + + assert.equal(getText(out.message), "the answer"); // worker turn still produced a draft + }); + + it("recovers from a failing role turn instead of crashing", async () => { + let n = 0; + const pool = { + w: { + id: "w", + async complete() { + n += 1; + if (n === 1) throw new Error("flaky"); + return { message: assistantMessage("recovered"), usage: { input: 1, output: 1 } }; + }, + }, + v: textModel('{"accepted":true}'), + }; + const triumvirat = createTriumviratModel({ + pool, + coordinator: scripted([ + { model: "w", role: "worker" }, + { model: "w", role: "worker" }, + { model: "v", role: "verifier" }, + ]), + }); + + const out = await triumvirat.complete(ask()); + + assert.equal(getText(out.message), "recovered"); + assert.equal(out.raw.accepted, true); + }); + + it("honors an aborted signal", async () => { + const pool = { w: textModel("x") }; + const controller = new AbortController(); + controller.abort(); + const triumvirat = createTriumviratModel({ pool, coordinator: scripted([{ model: "w", role: "worker" }]) }); + + await assert.rejects( + () => triumvirat.complete(ask(), { signal: controller.signal }), + /aborted/, + ); + }); + + it("validates configuration", () => { + assert.throws(() => createTriumviratModel({}), LlmConfigError); + assert.throws(() => createTriumviratModel({ pool: {} }), LlmConfigError); + assert.throws(() => createTriumviratModel({ pool: { a: { nope: 1 } } }), LlmConfigError); + assert.throws(() => createTriumviratModel({ pool: { a: textModel("x") }, maxTurns: 0 }), LlmConfigError); + }); +}); diff --git a/specs/current/10_core/requirements-matrix.md b/specs/current/10_core/requirements-matrix.md index 0c0ce698..6dbdac9e 100644 --- a/specs/current/10_core/requirements-matrix.md +++ b/specs/current/10_core/requirements-matrix.md @@ -14,7 +14,7 @@ when the weak/gap count rises against the baseline (`--ratchet`). |---|---|---| | UR-CORE-001 | ✅ | `packages/base-eval/tests/base-eval.test.mjs`
`packages/base-eval/tests/report.test.mjs`
`tests/base-context-pack.test.mjs`
`tests/base-core.test.mjs`
`tests/base-doctor.test.mjs`
`tests/base-egress.test.mjs`
`tests/base-feedback.test.mjs`
`tests/base-ontology.test.mjs`
`tests/base-perimeter.test.mjs`
`tests/browser-pack.test.mjs`
`tests/docs-canonical-phrase.test.mjs`
`tests/docs-tutorial.test.mjs`
`tests/eval-broker-harness.test.mjs`
`tests/eval-orchestrate.test.mjs`
`tests/eval-run.test.mjs`
`tests/eval-store.test.mjs`
`tests/eval-sut-tools.test.mjs`
`tests/studio-api.test.mjs`
`tests/studio-chat.test.mjs`
`tests/studio-eval.test.mjs`
`tests/studio-server.test.mjs`
`tests/studio-settings.test.mjs`
`tests/studio-watch.test.mjs` | | UR-CORE-002 | ✅ | `tests/base-cli-init.test.mjs`
`tests/base-cli.test.mjs`
`tests/base-userconfig.test.mjs` | -| UR-CORE-003 | ✅ | `packages/base-llm/tests/base-llm.streaming.test.mjs`
`packages/base-llm/tests/base-llm.test.mjs` | +| UR-CORE-003 | ✅ | `packages/base-llm/tests/base-llm.langfuse.test.mjs`
`packages/base-llm/tests/base-llm.moa.test.mjs`
`packages/base-llm/tests/base-llm.streaming.test.mjs`
`packages/base-llm/tests/base-llm.test.mjs`
`packages/base-llm/tests/base-llm.triumvirat.test.mjs`
`tests/model-settings-ensembles.test.mjs` | | UR-CORE-004 | ✅ | `tests/docs-contract.test.mjs`
`tests/requirements-matrix.test.mjs`
`tests/spec-gates.test.mjs` | | NFR-CORE-007 | ✅ | `tests/architecture.test.mjs`
`tests/base-core.test.mjs` | | NFR-CORE-008 | ✅ | `tests/base-core.test.mjs` | diff --git a/tests/model-settings-ensembles.test.mjs b/tests/model-settings-ensembles.test.mjs new file mode 100644 index 00000000..80fd4740 --- /dev/null +++ b/tests/model-settings-ensembles.test.mjs @@ -0,0 +1,83 @@ +// Spec coverage: UR-CORE-003 +import { describe, it, before, after } from "node:test"; +import assert from "node:assert/strict"; +import { mkdtemp, mkdir, writeFile, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import path from "node:path"; + +import { resolveModel } from "../tools/core/model-settings.mjs"; + +const PROVIDERS = [ + { id: "local", type: "ollama", baseUrl: "http://localhost:11434/v1" }, + { id: "local2", type: "ollama", baseUrl: "http://localhost:11435/v1" }, +]; + +async function makeContext(settings) { + const dir = await mkdtemp(path.join(tmpdir(), "base-ensembles-")); + await mkdir(path.join(dir, ".ai"), { recursive: true }); + await writeFile(path.join(dir, ".ai", "studio.settings.json"), JSON.stringify(settings), "utf8"); + return dir; +} + +describe("model-settings ensembles", () => { + let dir; + const trash = []; + const ctx = async (settings) => { + const d = await makeContext(settings); + trash.push(d); + return d; + }; + + before(async () => { + dir = await ctx({ + providers: PROVIDERS, + ensembles: { + council: { type: "moa", proposers: ["local/a", "local2/b"], aggregator: "local/a" }, + trio: { type: "triumvirat", pool: { x: "local/a", y: "local2/b" }, maxTurns: 4 }, + }, + }); + }); + after(async () => { + await Promise.all(trash.map((d) => rm(d, { recursive: true, force: true }))); + }); + + it("resolves a moa ensemble ref to a composed model with the ensemble's id", async () => { + const m = await resolveModel(dir, "council", { env: {} }); + assert.equal(m.id, "council"); + assert.equal(typeof m.complete, "function"); + }); + + it("resolves a triumvirat ensemble ref to a composed model", async () => { + const m = await resolveModel(dir, "trio", { env: {} }); + assert.equal(m.id, "trio"); + assert.equal(typeof m.complete, "function"); + }); + + it("still resolves a plain provider/model ref", async () => { + const m = await resolveModel(dir, "local/llama3", { env: {} }); + assert.equal(typeof m.complete, "function"); + }); + + it("rejects a moa ensemble missing its aggregator", async () => { + const d = await ctx({ providers: PROVIDERS, ensembles: { c: { type: "moa", proposers: ["local/a"] } } }); + await assert.rejects(() => resolveModel(d, "c", { env: {} }), /aggregator/); + }); + + it("rejects a triumvirat ensemble with an empty pool", async () => { + const d = await ctx({ providers: PROVIDERS, ensembles: { c: { type: "triumvirat", pool: {} } } }); + await assert.rejects(() => resolveModel(d, "c", { env: {} }), /pool/); + }); + + it("rejects an unknown ensemble type", async () => { + const d = await ctx({ providers: PROVIDERS, ensembles: { c: { type: "duo", proposers: [], aggregator: "x" } } }); + await assert.rejects(() => resolveModel(d, "c", { env: {} }), /unknown type/); + }); + + it("rejects an ensemble member pointing at an unknown provider", async () => { + const d = await ctx({ + providers: PROVIDERS, + ensembles: { c: { type: "moa", proposers: ["ghost/a"], aggregator: "local/a" } }, + }); + await assert.rejects(() => resolveModel(d, "c", { env: {} }), /unknown provider/); + }); +}); diff --git a/tests/studio-settings.test.mjs b/tests/studio-settings.test.mjs index 7009a54c..d85f48be 100644 --- a/tests/studio-settings.test.mjs +++ b/tests/studio-settings.test.mjs @@ -41,7 +41,7 @@ describe("studio settings — read/write", () => { it("reads empty settings when no file exists", async () => { const s = await readSettings(dir); - assert.deepEqual(s, { providers: [], aliases: {}, defaults: {}, discovered: {} }); + assert.deepEqual(s, { providers: [], aliases: {}, defaults: {}, discovered: {}, ensembles: {} }); }); it("writes valid settings and reads them back with keyDetected + locality, never the key", async () => { diff --git a/tools/core/model-settings.mjs b/tools/core/model-settings.mjs index 947df2c8..e19d728c 100644 --- a/tools/core/model-settings.mjs +++ b/tools/core/model-settings.mjs @@ -34,7 +34,7 @@ const KEY_REQUIRED_TYPES = new Set(["anthropic", "google"]); // The provider types that can embed (anthropic/google have no embeddings endpoint). const EMBEDDING_PROVIDER_TYPES = new Set(["ollama", "openai-compatible"]); -const EMPTY = { providers: [], aliases: {}, defaults: {}, discovered: {} }; +const EMPTY = { providers: [], aliases: {}, defaults: {}, discovered: {}, ensembles: {} }; const LOOPBACK = /^(localhost|127\.0\.0\.1|\[?::1\]?)$/i; /** @@ -92,10 +92,73 @@ export async function readSettings(contextDir, { env = process.env } = {}) { * read from the environment via the provider's apiKeyEnv. Unknown provider → BAD_REQUEST. */ export async function resolveModel(contextDir, ref, { env = process.env, fetch, timeoutMs } = /** @type {{ env?: NodeJS.ProcessEnv, fetch?: typeof globalThis.fetch, timeoutMs?: number }} */ ({})) { - const { provider, model } = await resolveProviderModel(contextDir, ref, { env }); + const settings = await readSettings(contextDir, { env }); + const name = String(ref ?? "").trim(); + // An ensemble ref (moa / triumvirat) resolves to a meta-model composed from member refs. + const ensemble = settings.ensembles?.[name]; + if (ensemble) return buildEnsembleModel(settings, name, ensemble, { env, fetch, timeoutMs }); + const { provider, model } = resolveProviderModelFrom(settings, ref); return await buildProviderModel(provider, model, { env, fetch, timeoutMs }); } +/** + * Build a MoA or Triumvirat meta-model from an ensemble settings block, resolving each member + * ref through the same provider registry as a normal model. + * @param {any} settings + * @param {string} name + * @param {any} ensemble + * @param {{ env?: any, fetch?: any, timeoutMs?: number }} opts + */ +async function buildEnsembleModel(settings, name, ensemble, { env = process.env, fetch, timeoutMs } = {}) { + const llm = await import("@ai-swiss/base-llm"); + const buildRef = async (ref) => { + const { provider, model } = resolveProviderModelFrom(settings, ref); + return buildProviderModel(provider, model, { env, fetch, timeoutMs }); + }; + const type = ensemble?.type; + + if (type === "moa") { + if (!Array.isArray(ensemble.proposers) || ensemble.proposers.length === 0) { + throw new ApiError(`ensemble "${name}" (moa) needs a non-empty "proposers" list`, "BAD_REQUEST"); + } + if (!ensemble.aggregator) { + throw new ApiError(`ensemble "${name}" (moa) needs an "aggregator" ref`, "BAD_REQUEST"); + } + const [proposers, aggregator] = await Promise.all([ + Promise.all(ensemble.proposers.map(buildRef)), + buildRef(ensemble.aggregator), + ]); + return llm.createMoaModel({ + proposers, + aggregator, + id: name, + ...(ensemble.synthesisPrompt ? { synthesisPrompt: ensemble.synthesisPrompt } : {}), + }); + } + + if (type === "triumvirat") { + const poolRefs = ensemble.pool; + if (!poolRefs || typeof poolRefs !== "object" || Array.isArray(poolRefs) || Object.keys(poolRefs).length === 0) { + throw new ApiError(`ensemble "${name}" (triumvirat) needs a non-empty "pool" object`, "BAD_REQUEST"); + } + const entries = await Promise.all( + Object.entries(poolRefs).map(async ([key, ref]) => [key, await buildRef(ref)]), + ); + const coordinator = ensemble.coordinator ? await buildRef(ensemble.coordinator) : undefined; + return llm.createTriumviratModel({ + pool: Object.fromEntries(entries), + id: name, + ...(coordinator ? { coordinator } : {}), + ...(Number.isInteger(ensemble.maxTurns) ? { maxTurns: ensemble.maxTurns } : {}), + }); + } + + throw new ApiError( + `ensemble "${name}" has unknown type "${type}" (expected "moa" or "triumvirat")`, + "BAD_REQUEST", + ); +} + /** * Resolve a `/` ref to its configured provider + model name — shared by resolveModel * and resolveEmbedder, so a chat model and an embedder of the same provider go through ONE registry and @@ -104,6 +167,11 @@ export async function resolveModel(contextDir, ref, { env = process.env, fetch, */ async function resolveProviderModel(contextDir, ref, { env = process.env } = {}) { const settings = await readSettings(contextDir, { env }); + return resolveProviderModelFrom(settings, ref); +} + +/** Resolve a ref against already-read settings (no file IO) — used by ensembles to resolve members. */ +function resolveProviderModelFrom(settings, ref) { const s = String(ref ?? "").trim(); const slash = s.indexOf("/"); let providerId;