mirror of
https://github.com/rowboatlabs/rowboat.git
synced 2026-07-12 21:02:17 +02:00
make local models work well
This commit is contained in:
Arjun
parent
596edcd788
commit
799d7584b8
23 changed files with 938 additions and 131 deletions
|
|
@ -528,7 +528,17 @@ function ModelSettings({ dialogOpen }: { dialogOpen: boolean }) {
|
|||
setDefaultProvider(provider)
|
||||
setTestState({ status: "success" })
|
||||
window.dispatchEvent(new Event('models-config-changed'))
|
||||
toast.success("Model configuration saved")
|
||||
// Capability probe caveats (local models): saved, but the user should
|
||||
// know when the model can't do tools or has a too-small context.
|
||||
const warnings: string[] = result.warnings ?? []
|
||||
if (warnings.length > 0) {
|
||||
for (const warning of warnings) {
|
||||
toast.warning(warning, { duration: 12000 })
|
||||
}
|
||||
toast.success("Model configuration saved (with warnings)")
|
||||
} else {
|
||||
toast.success("Model configuration saved")
|
||||
}
|
||||
} else {
|
||||
setTestState({ status: "error", error: result.error })
|
||||
toast.error(result.error || "Connection test failed")
|
||||
|
|
|
|||
|
|
@ -19,7 +19,7 @@ import { resolveFilePathForPermission } from "../filesystem/files.js";
|
|||
import container from "../di/container.js";
|
||||
import { notifyIfEnabled } from "../application/notification/notifier.js";
|
||||
import { IModelConfigRepo } from "../models/repo.js";
|
||||
import { createProvider } from "../models/models.js";
|
||||
import { createLanguageModel } from "../models/models.js";
|
||||
import { resolveProviderConfig } from "../models/defaults.js";
|
||||
import { IAgentsRepo } from "./repo.js";
|
||||
import { IMonotonicallyIncreasingIdGenerator } from "../application/lib/id-gen.js";
|
||||
|
|
@ -1365,8 +1365,8 @@ export async function* streamAgent({
|
|||
}
|
||||
const modelId = state.runModel;
|
||||
const providerConfig = await resolveProviderConfig(state.runProvider);
|
||||
const provider = createProvider(providerConfig);
|
||||
const model = provider.languageModel(modelId);
|
||||
// Legacy runs are user-facing chats: interactive priority on local models.
|
||||
const model = createLanguageModel(providerConfig, modelId, { priority: "interactive" });
|
||||
logger.log(`using model: ${modelId} (provider: ${state.runProvider})`);
|
||||
|
||||
// Install use-case context for tool-internal LLM calls (e.g. parseFile)
|
||||
|
|
|
|||
|
|
@ -93,7 +93,7 @@ async function resolveCodeProject(dirPath: string): Promise<
|
|||
import { ensureLoaded as ensureBrowserSkillsLoaded, readSkillContent as readBrowserSkillContent, refreshFromRemote as refreshBrowserSkills } from "../browser-skills/index.js";
|
||||
import type { ToolContext } from "./exec-tool.js";
|
||||
import { generateText } from "ai";
|
||||
import { createProvider } from "../../models/models.js";
|
||||
import { createLanguageModel } from "../../models/models.js";
|
||||
import { getDefaultModelAndProvider, resolveProviderConfig } from "../../models/defaults.js";
|
||||
import { captureLlmUsage } from "../../analytics/usage.js";
|
||||
import { getCurrentUseCase, withUseCase } from "../../analytics/use_case.js";
|
||||
|
|
@ -584,7 +584,8 @@ export const BuiltinTools: z.infer<typeof BuiltinToolsSchema> = {
|
|||
|
||||
const { model: modelId, provider: providerName } = await getDefaultModelAndProvider();
|
||||
const providerConfig = await resolveProviderConfig(providerName);
|
||||
const model = createProvider(providerConfig).languageModel(modelId);
|
||||
// Runs as a tool inside a chat turn more often than not.
|
||||
const model = createLanguageModel(providerConfig, modelId, { priority: 'interactive' });
|
||||
|
||||
const userPrompt = prompt || 'Convert this file to well-structured markdown.';
|
||||
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
import type { EventConsumer, EventConsumerTarget } from '../events/consumer.js';
|
||||
import { routeBatch } from '../events/routing.js';
|
||||
import { createProvider } from '../models/models.js';
|
||||
import { createLanguageModel } from '../models/models.js';
|
||||
import {
|
||||
getDefaultModelAndProvider,
|
||||
getBackgroundTaskAgentModel,
|
||||
|
|
@ -14,7 +14,7 @@ async function resolveRoutingModel() {
|
|||
const { provider } = await getDefaultModelAndProvider();
|
||||
const config = await resolveProviderConfig(provider);
|
||||
return {
|
||||
model: createProvider(config).languageModel(modelId),
|
||||
model: createLanguageModel(config, modelId, { priority: 'classifier' }),
|
||||
modelId,
|
||||
providerName: provider,
|
||||
};
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
import { generateObject } from 'ai';
|
||||
import type { LanguageModel } from 'ai';
|
||||
import { events, PrefixLogger } from '@x/shared';
|
||||
import { generateObjectSafe } from '../models/structured.js';
|
||||
import type { RowboatEvent } from '@x/shared/dist/events.js';
|
||||
import { captureLlmUsage } from '../analytics/usage.js';
|
||||
import { withUseCase, type UseCase } from '../analytics/use_case.js';
|
||||
|
|
@ -89,11 +89,12 @@ export async function routeBatch(
|
|||
for (let i = 0; i < targets.length; i += BATCH_SIZE) {
|
||||
const batch = targets.slice(i, i + BATCH_SIZE);
|
||||
try {
|
||||
const result = await withUseCase({ useCase: opts.useCase, subUseCase: 'routing' }, () => generateObject({
|
||||
const result = await withUseCase({ useCase: opts.useCase, subUseCase: 'routing' }, () => generateObjectSafe({
|
||||
model,
|
||||
system: systemPrompt,
|
||||
prompt: buildPrompt(event, batch, opts.entityPlural),
|
||||
schema: events.Pass1OutputSchema,
|
||||
retry: true,
|
||||
}));
|
||||
captureLlmUsage({
|
||||
useCase: opts.useCase,
|
||||
|
|
|
|||
|
|
@ -1,11 +1,11 @@
|
|||
import fs from 'fs';
|
||||
import path from 'path';
|
||||
import { z } from 'zod';
|
||||
import { generateObject } from 'ai';
|
||||
import { google } from 'googleapis';
|
||||
import type { OAuth2Client } from 'google-auth-library';
|
||||
import { WorkDir } from '../config/config.js';
|
||||
import { createProvider } from '../models/models.js';
|
||||
import { createLanguageModel } from '../models/models.js';
|
||||
import { generateObjectSafe } from '../models/structured.js';
|
||||
import {
|
||||
getDefaultModelAndProvider,
|
||||
getKgModel,
|
||||
|
|
@ -249,7 +249,7 @@ export async function classifyThread(
|
|||
const modelId = await getKgModel();
|
||||
const { provider } = await getDefaultModelAndProvider();
|
||||
const config = await resolveProviderConfig(provider);
|
||||
const model = createProvider(config).languageModel(modelId);
|
||||
const model = createLanguageModel(config, modelId, { priority: 'background' });
|
||||
|
||||
let systemPrompt = options.skipDraft
|
||||
? `${SYSTEM_PROMPT}\n\n# Skip the draft\n\nThe user already has their own draft in progress for this thread — DO NOT generate a draftResponse. Always omit the draftResponse field.`
|
||||
|
|
@ -262,11 +262,12 @@ export async function classifyThread(
|
|||
systemPrompt = `${systemPrompt}\n\n${feedback}`;
|
||||
}
|
||||
|
||||
const result = await withUseCase({ useCase: 'knowledge_sync', subUseCase: 'email_classifier' }, () => generateObject({
|
||||
const result = await withUseCase({ useCase: 'knowledge_sync', subUseCase: 'email_classifier' }, () => generateObjectSafe({
|
||||
model,
|
||||
system: systemPrompt,
|
||||
prompt: buildPrompt(snapshot, userEmail, styleGuide, calendar),
|
||||
schema: ClassificationSchema,
|
||||
retry: true,
|
||||
}));
|
||||
|
||||
captureLlmUsage({
|
||||
|
|
|
|||
|
|
@ -7,7 +7,7 @@ import { runHeadlessAgent } from '../agents/headless-app.js';
|
|||
import { getKgModel } from '../models/defaults.js';
|
||||
import container from '../di/container.js';
|
||||
import type { IModelConfigRepo } from '../models/repo.js';
|
||||
import { createProvider } from '../models/models.js';
|
||||
import { createLanguageModel } from '../models/models.js';
|
||||
import { inlineTask } from '@x/shared';
|
||||
import { captureLlmUsage } from '../analytics/usage.js';
|
||||
import { withUseCase } from '../analytics/use_case.js';
|
||||
|
|
@ -613,8 +613,7 @@ export async function processRowboatInstruction(
|
|||
export async function classifySchedule(instruction: string): Promise<InlineTaskSchedule | null> {
|
||||
const repo = container.resolve<IModelConfigRepo>('modelConfigRepo');
|
||||
const config = await repo.getConfig();
|
||||
const provider = createProvider(config.provider);
|
||||
const model = provider.languageModel(config.model);
|
||||
const model = createLanguageModel(config.provider, config.model, { priority: 'classifier' });
|
||||
|
||||
const now = new Date();
|
||||
const defaultEnd = new Date(now.getTime() + 7 * 24 * 60 * 60 * 1000);
|
||||
|
|
|
|||
|
|
@ -3,7 +3,7 @@ import { fetchLiveNote } from './fileops.js';
|
|||
import { runLiveNoteAgent } from './runner.js';
|
||||
import type { EventConsumer, EventConsumerTarget } from '../../events/consumer.js';
|
||||
import { routeBatch } from '../../events/routing.js';
|
||||
import { createProvider } from '../../models/models.js';
|
||||
import { createLanguageModel } from '../../models/models.js';
|
||||
import { getDefaultModelAndProvider, getLiveNoteAgentModel, resolveProviderConfig } from '../../models/defaults.js';
|
||||
|
||||
async function resolveRoutingModel() {
|
||||
|
|
@ -11,7 +11,7 @@ async function resolveRoutingModel() {
|
|||
const { provider } = await getDefaultModelAndProvider();
|
||||
const config = await resolveProviderConfig(provider);
|
||||
return {
|
||||
model: createProvider(config).languageModel(modelId),
|
||||
model: createLanguageModel(config, modelId, { priority: 'classifier' }),
|
||||
modelId,
|
||||
providerName: provider,
|
||||
};
|
||||
|
|
|
|||
|
|
@ -2,7 +2,7 @@ import fs from 'node:fs/promises';
|
|||
import path from 'node:path';
|
||||
import { generateText } from 'ai';
|
||||
import { WorkDir } from '../config/config.js';
|
||||
import { createProvider } from '../models/models.js';
|
||||
import { createLanguageModel } from '../models/models.js';
|
||||
import { getDefaultModelAndProvider, getMeetingNotesModel, resolveProviderConfig } from '../models/defaults.js';
|
||||
import { captureLlmUsage } from '../analytics/usage.js';
|
||||
import { withUseCase } from '../analytics/use_case.js';
|
||||
|
|
@ -178,7 +178,7 @@ async function generateBrief(event: CalendarEvent, ctx: Awaited<ReturnType<typeo
|
|||
const modelId = await getMeetingNotesModel();
|
||||
const { provider: providerName } = await getDefaultModelAndProvider();
|
||||
const providerConfig = await resolveProviderConfig(providerName);
|
||||
const model = createProvider(providerConfig).languageModel(modelId);
|
||||
const model = createLanguageModel(providerConfig, modelId, { priority: 'background' });
|
||||
|
||||
const result = await withUseCase({ useCase: 'meeting_prep' }, () => generateText({
|
||||
model,
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
import fs from 'fs';
|
||||
import path from 'path';
|
||||
import { generateText } from 'ai';
|
||||
import { createProvider } from '../models/models.js';
|
||||
import { createLanguageModel } from '../models/models.js';
|
||||
import { getDefaultModelAndProvider, getMeetingNotesModel, resolveProviderConfig } from '../models/defaults.js';
|
||||
import { WorkDir } from '../config/config.js';
|
||||
import { captureLlmUsage } from '../analytics/usage.js';
|
||||
|
|
@ -140,7 +140,7 @@ export async function summarizeMeeting(transcript: string, meetingStartTime?: st
|
|||
const modelId = await getMeetingNotesModel();
|
||||
const { provider: providerName } = await getDefaultModelAndProvider();
|
||||
const providerConfig = await resolveProviderConfig(providerName);
|
||||
const model = createProvider(providerConfig).languageModel(modelId);
|
||||
const model = createLanguageModel(providerConfig, modelId, { priority: 'background' });
|
||||
|
||||
// If a specific calendar event was linked, use it directly.
|
||||
// Otherwise fall back to scanning events within ±3 hours.
|
||||
|
|
|
|||
|
|
@ -52,6 +52,7 @@ export async function resolveProviderConfig(name: string): Promise<z.infer<typeo
|
|||
apiKey: entry.apiKey,
|
||||
baseURL: entry.baseURL,
|
||||
headers: entry.headers,
|
||||
contextLength: entry.contextLength,
|
||||
});
|
||||
}
|
||||
if (cfg.provider.flavor === name) {
|
||||
|
|
|
|||
118
apps/x/packages/core/src/models/local.test.ts
Normal file
118
apps/x/packages/core/src/models/local.test.ts
Normal file
|
|
@ -0,0 +1,118 @@
|
|||
import { describe, expect, it } from "vitest";
|
||||
import { LocalLlmScheduler, isLocalProvider } from "./local.js";
|
||||
|
||||
function deferred() {
|
||||
let resolve!: () => void;
|
||||
const promise = new Promise<void>((r) => {
|
||||
resolve = r;
|
||||
});
|
||||
return { promise, resolve };
|
||||
}
|
||||
|
||||
const tick = () => new Promise<void>((r) => setTimeout(r, 0));
|
||||
|
||||
describe("isLocalProvider", () => {
|
||||
it("treats ollama as local", () => {
|
||||
expect(isLocalProvider({ flavor: "ollama" })).toBe(true);
|
||||
});
|
||||
|
||||
it("treats loopback openai-compatible endpoints as local", () => {
|
||||
expect(isLocalProvider({ flavor: "openai-compatible", baseURL: "http://localhost:1234/v1" })).toBe(true);
|
||||
expect(isLocalProvider({ flavor: "openai-compatible", baseURL: "http://127.0.0.1:8080/v1" })).toBe(true);
|
||||
});
|
||||
|
||||
it("treats remote openai-compatible endpoints and cloud flavors as non-local", () => {
|
||||
expect(isLocalProvider({ flavor: "openai-compatible", baseURL: "https://api.together.xyz/v1" })).toBe(false);
|
||||
expect(isLocalProvider({ flavor: "openai" })).toBe(false);
|
||||
expect(isLocalProvider({ flavor: "rowboat" })).toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
describe("LocalLlmScheduler", () => {
|
||||
it("serves waiters by priority, FIFO within a priority", async () => {
|
||||
const scheduler = new LocalLlmScheduler(1);
|
||||
const order: string[] = [];
|
||||
const first = deferred();
|
||||
|
||||
const running = scheduler.run("background", undefined, async () => {
|
||||
await first.promise;
|
||||
order.push("initial");
|
||||
});
|
||||
await tick();
|
||||
|
||||
const bg1 = scheduler.run("background", undefined, async () => {
|
||||
order.push("bg1");
|
||||
});
|
||||
const bg2 = scheduler.run("background", undefined, async () => {
|
||||
order.push("bg2");
|
||||
});
|
||||
const chat = scheduler.run("interactive", undefined, async () => {
|
||||
order.push("chat");
|
||||
});
|
||||
const classifier = scheduler.run("classifier", undefined, async () => {
|
||||
order.push("classifier");
|
||||
});
|
||||
await tick();
|
||||
|
||||
first.resolve();
|
||||
await Promise.all([running, bg1, bg2, chat, classifier]);
|
||||
expect(order).toEqual(["initial", "chat", "classifier", "bg1", "bg2"]);
|
||||
});
|
||||
|
||||
it("releases the slot when a task throws", async () => {
|
||||
const scheduler = new LocalLlmScheduler(1);
|
||||
await expect(
|
||||
scheduler.run("interactive", undefined, async () => {
|
||||
throw new Error("boom");
|
||||
}),
|
||||
).rejects.toThrow("boom");
|
||||
// Slot must be free again.
|
||||
await scheduler.run("interactive", undefined, async () => undefined);
|
||||
});
|
||||
|
||||
it("rejects queued waiters whose signal aborts, without leaking the slot", async () => {
|
||||
const scheduler = new LocalLlmScheduler(1);
|
||||
const gate = deferred();
|
||||
const running = scheduler.run("background", undefined, () => gate.promise);
|
||||
await tick();
|
||||
|
||||
const controller = new AbortController();
|
||||
const waiting = scheduler.acquire("interactive", controller.signal);
|
||||
controller.abort();
|
||||
await expect(waiting).rejects.toThrow(/abort/i);
|
||||
|
||||
gate.resolve();
|
||||
await running;
|
||||
// Queue is clean: a fresh acquire succeeds immediately.
|
||||
const release = await scheduler.acquire("background");
|
||||
release();
|
||||
});
|
||||
|
||||
it("rejects immediately when acquiring with an already-aborted signal", async () => {
|
||||
const scheduler = new LocalLlmScheduler(1);
|
||||
const controller = new AbortController();
|
||||
controller.abort();
|
||||
await expect(scheduler.acquire("interactive", controller.signal)).rejects.toThrow(/abort/i);
|
||||
});
|
||||
|
||||
it("allows up to maxConcurrent tasks at once", async () => {
|
||||
const scheduler = new LocalLlmScheduler(2);
|
||||
const gateA = deferred();
|
||||
const gateB = deferred();
|
||||
let active = 0;
|
||||
let peak = 0;
|
||||
const track = async (gate: Promise<void>) => {
|
||||
active++;
|
||||
peak = Math.max(peak, active);
|
||||
await gate;
|
||||
active--;
|
||||
};
|
||||
const a = scheduler.run("background", undefined, () => track(gateA.promise));
|
||||
const b = scheduler.run("background", undefined, () => track(gateB.promise));
|
||||
await tick();
|
||||
expect(peak).toBe(2);
|
||||
gateA.resolve();
|
||||
gateB.resolve();
|
||||
await Promise.all([a, b]);
|
||||
});
|
||||
});
|
||||
252
apps/x/packages/core/src/models/local.ts
Normal file
252
apps/x/packages/core/src/models/local.ts
Normal file
|
|
@ -0,0 +1,252 @@
|
|||
import { wrapLanguageModel, type LanguageModel } from "ai";
|
||||
import type { z } from "zod";
|
||||
import type { LlmProvider } from "@x/shared/dist/models.js";
|
||||
import { PrefixLogger } from "@x/shared";
|
||||
|
||||
const log = new PrefixLogger("LocalLlm");
|
||||
|
||||
// Ollama's server-side default context window (~4k tokens) is far below what
|
||||
// Rowboat's agents need (the copilot's system prompt + tool schemas alone are
|
||||
// ~15-20k tokens) and Ollama silently truncates the prompt from the top when
|
||||
// it overflows — the model loses its own instructions. We therefore always
|
||||
// request an explicit window for Ollama models. Overridable per provider via
|
||||
// `contextLength` in models.json.
|
||||
export const DEFAULT_OLLAMA_CONTEXT_LENGTH = 32768;
|
||||
|
||||
const LOOPBACK_HOSTS = new Set(["localhost", "127.0.0.1", "0.0.0.0", "::1", "[::1]"]);
|
||||
|
||||
/**
|
||||
* Whether requests to this provider are served by a model running on the
|
||||
* user's own machine (and therefore need scheduling: local runtimes process
|
||||
* requests mostly serially, so background work must not starve chat).
|
||||
*/
|
||||
export function isLocalProvider(config: z.infer<typeof LlmProvider>): boolean {
|
||||
if (config.flavor === "ollama") {
|
||||
return true;
|
||||
}
|
||||
if (config.flavor === "openai-compatible") {
|
||||
// LM Studio, llama.cpp server, vLLM on the same machine, etc.
|
||||
try {
|
||||
const host = new URL(config.baseURL ?? "").hostname;
|
||||
return LOOPBACK_HOSTS.has(host) || host.endsWith(".local");
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
export type LlmPriority = "interactive" | "classifier" | "background";
|
||||
|
||||
const PRIORITY_ORDER: Record<LlmPriority, number> = {
|
||||
interactive: 0,
|
||||
classifier: 1,
|
||||
background: 2,
|
||||
};
|
||||
|
||||
interface Waiter {
|
||||
priority: LlmPriority;
|
||||
seq: number;
|
||||
resolve: (release: () => void) => void;
|
||||
reject: (error: Error) => void;
|
||||
signal?: AbortSignal;
|
||||
onAbort?: () => void;
|
||||
}
|
||||
|
||||
function abortError(): Error {
|
||||
const error = new Error("local LLM slot acquisition aborted");
|
||||
error.name = "AbortError";
|
||||
return error;
|
||||
}
|
||||
|
||||
/**
|
||||
* Serializes access to local LLM runtimes. One slot by default: local servers
|
||||
* effectively process one request at a time, and interleaving requests
|
||||
* destroys their KV-cache reuse. Waiters are served strictly by priority
|
||||
* (interactive chat > lightweight classifiers > background knowledge
|
||||
* pipeline), FIFO within a priority — so a queued email-labeling job can
|
||||
* never delay the user's chat by more than the one request already running.
|
||||
*/
|
||||
export class LocalLlmScheduler {
|
||||
private active = 0;
|
||||
private seq = 0;
|
||||
private readonly waiting: Waiter[] = [];
|
||||
|
||||
constructor(private readonly maxConcurrent = 1) {}
|
||||
|
||||
async acquire(priority: LlmPriority, signal?: AbortSignal): Promise<() => void> {
|
||||
if (signal?.aborted) {
|
||||
throw abortError();
|
||||
}
|
||||
if (this.active < this.maxConcurrent) {
|
||||
this.active++;
|
||||
return this.makeRelease();
|
||||
}
|
||||
return new Promise<() => void>((resolve, reject) => {
|
||||
const waiter: Waiter = { priority, seq: this.seq++, resolve, reject, signal };
|
||||
if (signal) {
|
||||
waiter.onAbort = () => {
|
||||
const index = this.waiting.indexOf(waiter);
|
||||
if (index >= 0) {
|
||||
this.waiting.splice(index, 1);
|
||||
reject(abortError());
|
||||
}
|
||||
};
|
||||
signal.addEventListener("abort", waiter.onAbort, { once: true });
|
||||
}
|
||||
this.waiting.push(waiter);
|
||||
if (this.waiting.length === 1 || this.waiting.length % 10 === 0) {
|
||||
log.log(`queueing ${priority} request (${this.waiting.length} waiting, ${this.active} active)`);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/** Run `fn` while holding a slot. */
|
||||
async run<T>(priority: LlmPriority, signal: AbortSignal | undefined, fn: () => PromiseLike<T>): Promise<T> {
|
||||
const release = await this.acquire(priority, signal);
|
||||
try {
|
||||
return await fn();
|
||||
} finally {
|
||||
release();
|
||||
}
|
||||
}
|
||||
|
||||
get queueDepth(): number {
|
||||
return this.waiting.length;
|
||||
}
|
||||
|
||||
private makeRelease(): () => void {
|
||||
let released = false;
|
||||
return () => {
|
||||
if (released) {
|
||||
return;
|
||||
}
|
||||
released = true;
|
||||
this.active--;
|
||||
this.dequeue();
|
||||
};
|
||||
}
|
||||
|
||||
private dequeue(): void {
|
||||
while (this.active < this.maxConcurrent && this.waiting.length > 0) {
|
||||
let best = 0;
|
||||
for (let i = 1; i < this.waiting.length; i++) {
|
||||
const a = this.waiting[i];
|
||||
const b = this.waiting[best];
|
||||
if (
|
||||
PRIORITY_ORDER[a.priority] < PRIORITY_ORDER[b.priority] ||
|
||||
(PRIORITY_ORDER[a.priority] === PRIORITY_ORDER[b.priority] && a.seq < b.seq)
|
||||
) {
|
||||
best = i;
|
||||
}
|
||||
}
|
||||
const [waiter] = this.waiting.splice(best, 1);
|
||||
if (waiter.signal && waiter.onAbort) {
|
||||
waiter.signal.removeEventListener("abort", waiter.onAbort);
|
||||
}
|
||||
this.active++;
|
||||
waiter.resolve(this.makeRelease());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function envConcurrency(): number {
|
||||
const raw = process.env.ROWBOAT_LOCAL_LLM_CONCURRENCY;
|
||||
const parsed = raw ? Number.parseInt(raw, 10) : NaN;
|
||||
return Number.isFinite(parsed) && parsed > 0 ? parsed : 1;
|
||||
}
|
||||
|
||||
// One queue for all local providers: users run one local server, and even
|
||||
// with several, the machine's compute is the shared resource.
|
||||
export const localLlmScheduler = new LocalLlmScheduler(envConcurrency());
|
||||
|
||||
/**
|
||||
* Wrap a language model so every call requests an explicit context window
|
||||
* from Ollama (merged under the caller's providerOptions — an explicit
|
||||
* caller value wins) and, when `priority` is set and the provider is local,
|
||||
* goes through the shared scheduler.
|
||||
*/
|
||||
export function applyLocalModelSettings(
|
||||
model: LanguageModel,
|
||||
providerConfig: z.infer<typeof LlmProvider>,
|
||||
priority: LlmPriority | null,
|
||||
): LanguageModel {
|
||||
if (typeof model === "string") {
|
||||
// Bare model-id strings resolve through the global registry; local
|
||||
// providers never take this path.
|
||||
return model;
|
||||
}
|
||||
const local = isLocalProvider(providerConfig);
|
||||
const wantsNumCtx = providerConfig.flavor === "ollama";
|
||||
if (!wantsNumCtx && !(local && priority)) {
|
||||
return model;
|
||||
}
|
||||
const numCtx = providerConfig.contextLength ?? DEFAULT_OLLAMA_CONTEXT_LENGTH;
|
||||
const schedule = local && priority ? priority : null;
|
||||
return wrapLanguageModel({
|
||||
model,
|
||||
middleware: {
|
||||
...(wantsNumCtx
|
||||
? {
|
||||
transformParams: async ({ params }) => {
|
||||
const providerOptions = (params.providerOptions ?? {}) as Record<string, Record<string, unknown>>;
|
||||
const ollama = (providerOptions.ollama ?? {}) as Record<string, unknown>;
|
||||
const options = (ollama.options ?? {}) as Record<string, unknown>;
|
||||
return {
|
||||
...params,
|
||||
providerOptions: {
|
||||
...providerOptions,
|
||||
ollama: {
|
||||
...ollama,
|
||||
options: { num_ctx: numCtx, ...options },
|
||||
},
|
||||
},
|
||||
};
|
||||
},
|
||||
}
|
||||
: {}),
|
||||
...(schedule
|
||||
? {
|
||||
wrapGenerate: async ({ doGenerate, params }) =>
|
||||
localLlmScheduler.run(schedule, params.abortSignal, () => doGenerate()),
|
||||
wrapStream: async ({ doStream, params }) => {
|
||||
const release = await localLlmScheduler.acquire(schedule, params.abortSignal);
|
||||
try {
|
||||
const { stream, ...rest } = await doStream();
|
||||
return { ...rest, stream: releaseOnSettled(stream, release) };
|
||||
} catch (error) {
|
||||
release();
|
||||
throw error;
|
||||
}
|
||||
},
|
||||
}
|
||||
: {}),
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
// Hold the slot until the provider stream drains, errors, or is cancelled —
|
||||
// a streaming response occupies the local server for its full duration.
|
||||
function releaseOnSettled<T>(stream: ReadableStream<T>, release: () => void): ReadableStream<T> {
|
||||
const reader = stream.getReader();
|
||||
return new ReadableStream<T>({
|
||||
async pull(controller) {
|
||||
try {
|
||||
const { done, value } = await reader.read();
|
||||
if (done) {
|
||||
release();
|
||||
controller.close();
|
||||
} else {
|
||||
controller.enqueue(value);
|
||||
}
|
||||
} catch (error) {
|
||||
release();
|
||||
controller.error(error);
|
||||
}
|
||||
},
|
||||
cancel(reason) {
|
||||
release();
|
||||
return reader.cancel(reason);
|
||||
},
|
||||
});
|
||||
}
|
||||
|
|
@ -1,5 +1,5 @@
|
|||
import { ProviderV2 } from "@ai-sdk/provider";
|
||||
import { createGateway, generateText } from "ai";
|
||||
import { createGateway, generateText, type LanguageModel } from "ai";
|
||||
import { createOpenAI } from "@ai-sdk/openai";
|
||||
import { createGoogleGenerativeAI } from "@ai-sdk/google";
|
||||
import { createAnthropic } from "@ai-sdk/anthropic";
|
||||
|
|
@ -12,6 +12,12 @@ import { getGatewayProvider } from "./gateway.js";
|
|||
import { getDefaultModelAndProvider, resolveProviderConfig } from "./defaults.js";
|
||||
import { getChatModelIds } from "./models-dev.js";
|
||||
import { withUseCase } from "../analytics/use_case.js";
|
||||
import {
|
||||
applyLocalModelSettings,
|
||||
isLocalProvider,
|
||||
DEFAULT_OLLAMA_CONTEXT_LENGTH,
|
||||
type LlmPriority,
|
||||
} from "./local.js";
|
||||
|
||||
export const Provider = LlmProvider;
|
||||
export const ModelConfig = LlmModelConfig;
|
||||
|
|
@ -74,24 +80,149 @@ export function createProvider(config: z.infer<typeof Provider>): ProviderV2 {
|
|||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* The one place model instances are created. Applies local-runtime settings
|
||||
* (explicit Ollama context window) and, when `priority` is given, routes the
|
||||
* call through the shared local scheduler so background work cannot starve
|
||||
* interactive chat. Pass `priority: null` when the caller does its own
|
||||
* scheduling (the turn runtime's model registry).
|
||||
*/
|
||||
export function createLanguageModel(
|
||||
providerConfig: z.infer<typeof Provider>,
|
||||
modelId: string,
|
||||
opts: { priority?: LlmPriority | null } = {},
|
||||
): LanguageModel {
|
||||
const model = createProvider(providerConfig).languageModel(modelId);
|
||||
return applyLocalModelSettings(model, providerConfig, opts.priority ?? null);
|
||||
}
|
||||
|
||||
export interface ModelCapabilities {
|
||||
/** undefined = could not be determined (endpoint missing, non-local provider). */
|
||||
supportsTools?: boolean;
|
||||
maxContextLength?: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* Best-effort capability probe for local runtimes. Ollama reports a
|
||||
* `capabilities` list and the model's trained context window via /api/show;
|
||||
* LM Studio exposes the same through its /api/v0/models REST endpoint.
|
||||
* Failures are swallowed — an unknown capability is not an error.
|
||||
*/
|
||||
export async function probeModelCapabilities(
|
||||
providerConfig: z.infer<typeof Provider>,
|
||||
model: string,
|
||||
timeoutMs = 5000,
|
||||
): Promise<ModelCapabilities> {
|
||||
const controller = new AbortController();
|
||||
const timeout = setTimeout(() => controller.abort(), timeoutMs);
|
||||
try {
|
||||
if (providerConfig.flavor === "ollama") {
|
||||
const base = (providerConfig.baseURL ?? "http://localhost:11434")
|
||||
.replace(/\/+$/, "")
|
||||
.replace(/\/api$/, "");
|
||||
const res = await fetch(`${base}/api/show`, {
|
||||
method: "POST",
|
||||
headers: { "Content-Type": "application/json", ...(providerConfig.headers ?? {}) },
|
||||
body: JSON.stringify({ model }),
|
||||
signal: controller.signal,
|
||||
});
|
||||
if (!res.ok) return {};
|
||||
const data = await res.json() as {
|
||||
capabilities?: string[];
|
||||
model_info?: Record<string, unknown>;
|
||||
};
|
||||
const result: ModelCapabilities = {};
|
||||
if (Array.isArray(data.capabilities)) {
|
||||
result.supportsTools = data.capabilities.includes("tools");
|
||||
}
|
||||
for (const [key, value] of Object.entries(data.model_info ?? {})) {
|
||||
if (key.endsWith(".context_length") && typeof value === "number") {
|
||||
result.maxContextLength = value;
|
||||
break;
|
||||
}
|
||||
}
|
||||
return result;
|
||||
}
|
||||
if (providerConfig.flavor === "openai-compatible" && isLocalProvider(providerConfig)) {
|
||||
// LM Studio's enhanced REST API lives at /api/v0 on the same
|
||||
// origin as the OpenAI-compatible /v1 endpoint.
|
||||
const origin = new URL(providerConfig.baseURL ?? "").origin;
|
||||
const res = await fetch(`${origin}/api/v0/models`, {
|
||||
headers: providerConfig.headers ?? {},
|
||||
signal: controller.signal,
|
||||
});
|
||||
if (!res.ok) return {};
|
||||
const data = await res.json() as { data?: Array<Record<string, unknown>> };
|
||||
const entry = (data.data ?? []).find((m) => m.id === model);
|
||||
if (!entry) return {};
|
||||
const result: ModelCapabilities = {};
|
||||
if (Array.isArray(entry.capabilities)) {
|
||||
result.supportsTools = (entry.capabilities as string[]).includes("tool_use");
|
||||
}
|
||||
const max = entry.loaded_context_length ?? entry.max_context_length;
|
||||
if (typeof max === "number") {
|
||||
result.maxContextLength = max;
|
||||
}
|
||||
return result;
|
||||
}
|
||||
return {};
|
||||
} catch {
|
||||
return {};
|
||||
} finally {
|
||||
clearTimeout(timeout);
|
||||
}
|
||||
}
|
||||
|
||||
function capabilityWarnings(
|
||||
providerConfig: z.infer<typeof Provider>,
|
||||
model: string,
|
||||
capabilities: ModelCapabilities,
|
||||
): string[] {
|
||||
const warnings: string[] = [];
|
||||
if (capabilities.supportsTools === false) {
|
||||
warnings.push(
|
||||
`${model} does not support tool calling. Rowboat's assistant and background agents rely on tools; pick a tool-capable model (e.g. qwen3, gpt-oss, llama3.3).`,
|
||||
);
|
||||
}
|
||||
const configured = providerConfig.contextLength
|
||||
?? (providerConfig.flavor === "ollama" ? DEFAULT_OLLAMA_CONTEXT_LENGTH : undefined);
|
||||
if (capabilities.maxContextLength !== undefined) {
|
||||
if (capabilities.maxContextLength < 16384) {
|
||||
warnings.push(
|
||||
`${model} has a ${capabilities.maxContextLength}-token context window. Rowboat's assistant needs ~16k+ tokens; expect truncated or confused responses.`,
|
||||
);
|
||||
} else if (configured !== undefined && capabilities.maxContextLength < configured) {
|
||||
warnings.push(
|
||||
`${model} supports at most ${capabilities.maxContextLength} context tokens, below the configured ${configured}. Set "contextLength" for this provider in models.json to ${capabilities.maxContextLength} or less.`,
|
||||
);
|
||||
}
|
||||
}
|
||||
return warnings;
|
||||
}
|
||||
|
||||
export async function testModelConnection(
|
||||
providerConfig: z.infer<typeof Provider>,
|
||||
model: string,
|
||||
timeoutMs?: number,
|
||||
): Promise<{ success: boolean; error?: string }> {
|
||||
): Promise<{ success: boolean; error?: string; warnings?: string[]; capabilities?: ModelCapabilities }> {
|
||||
const isLocal = providerConfig.flavor === "ollama" || providerConfig.flavor === "openai-compatible";
|
||||
const effectiveTimeout = timeoutMs ?? (isLocal ? 60000 : 8000);
|
||||
const controller = new AbortController();
|
||||
const timeout = setTimeout(() => controller.abort(), effectiveTimeout);
|
||||
try {
|
||||
const provider = createProvider(providerConfig);
|
||||
const languageModel = provider.languageModel(model);
|
||||
const languageModel = createLanguageModel(providerConfig, model);
|
||||
await generateText({
|
||||
model: languageModel,
|
||||
prompt: "ping",
|
||||
abortSignal: controller.signal,
|
||||
});
|
||||
return { success: true };
|
||||
const capabilities = await probeModelCapabilities(providerConfig, model);
|
||||
const warnings = capabilityWarnings(providerConfig, model, capabilities);
|
||||
return {
|
||||
success: true,
|
||||
...(warnings.length > 0 ? { warnings } : {}),
|
||||
capabilities,
|
||||
};
|
||||
} catch (error) {
|
||||
const message = error instanceof Error ? error.message : "Connection test failed";
|
||||
return { success: false, error: message };
|
||||
|
|
@ -203,7 +334,7 @@ export async function generateOneShot(opts: GenerateTextOptions): Promise<Genera
|
|||
const modelId = opts.model || def.model;
|
||||
const providerName = opts.provider || def.provider;
|
||||
const providerConfig = await resolveProviderConfig(providerName);
|
||||
const languageModel = createProvider(providerConfig).languageModel(modelId);
|
||||
const languageModel = createLanguageModel(providerConfig, modelId, { priority: "interactive" });
|
||||
const result = await withUseCase(
|
||||
{ useCase: "copilot_chat", subUseCase: "email_compose" },
|
||||
() => generateText({
|
||||
|
|
|
|||
|
|
@ -48,6 +48,10 @@ export class FSModelConfigRepo implements IModelConfigRepo {
|
|||
apiKey: config.provider.apiKey,
|
||||
baseURL: config.provider.baseURL,
|
||||
headers: config.provider.headers,
|
||||
// Preserve a hand-edited contextLength unless the caller sets one.
|
||||
...(config.provider.contextLength !== undefined
|
||||
? { contextLength: config.provider.contextLength }
|
||||
: {}),
|
||||
model: config.model,
|
||||
models: config.models,
|
||||
knowledgeGraphModel: config.knowledgeGraphModel,
|
||||
|
|
|
|||
87
apps/x/packages/core/src/models/structured.test.ts
Normal file
87
apps/x/packages/core/src/models/structured.test.ts
Normal file
|
|
@ -0,0 +1,87 @@
|
|||
import { describe, expect, it } from "vitest";
|
||||
import { z } from "zod";
|
||||
import { NoObjectGeneratedError } from "ai";
|
||||
import type { LanguageModel } from "ai";
|
||||
import { generateObjectSafe } from "./structured.js";
|
||||
|
||||
const Schema = z.object({ ids: z.array(z.string()) });
|
||||
|
||||
// A minimal LanguageModelV2 double whose doGenerate returns the given texts
|
||||
// in sequence. generateObject parses the text against the schema itself, so
|
||||
// malformed text surfaces as NoObjectGeneratedError — exactly the local-model
|
||||
// failure mode generateObjectSafe exists to absorb.
|
||||
function fakeModel(responses: string[]): LanguageModel {
|
||||
let call = 0;
|
||||
return {
|
||||
specificationVersion: "v2",
|
||||
provider: "fake",
|
||||
modelId: "fake-model",
|
||||
supportedUrls: {},
|
||||
doGenerate: async () => {
|
||||
const text = responses[Math.min(call++, responses.length - 1)];
|
||||
return {
|
||||
content: [{ type: "text" as const, text }],
|
||||
finishReason: "stop" as const,
|
||||
usage: { inputTokens: 1, outputTokens: 1, totalTokens: 2 },
|
||||
warnings: [],
|
||||
};
|
||||
},
|
||||
doStream: async () => {
|
||||
throw new Error("not used");
|
||||
},
|
||||
} as unknown as LanguageModel;
|
||||
}
|
||||
|
||||
describe("generateObjectSafe", () => {
|
||||
it("passes through a clean structured response", async () => {
|
||||
const result = await generateObjectSafe({
|
||||
model: fakeModel(['{"ids":["a","b"]}']),
|
||||
prompt: "p",
|
||||
schema: Schema,
|
||||
});
|
||||
expect(result.object).toEqual({ ids: ["a", "b"] });
|
||||
expect(result.usage?.totalTokens).toBe(2);
|
||||
});
|
||||
|
||||
it("salvages JSON wrapped in prose and think blocks", async () => {
|
||||
const result = await generateObjectSafe({
|
||||
model: fakeModel([
|
||||
'<think>hmm {"ids":["x"]} maybe</think>Sure! Here you go:\n```json\n{"ids":["note-1","note-2"]}\n```\nLet me know!',
|
||||
]),
|
||||
prompt: "p",
|
||||
schema: Schema,
|
||||
});
|
||||
expect(result.object).toEqual({ ids: ["note-1", "note-2"] });
|
||||
});
|
||||
|
||||
it("retries once with a reinforced instruction when enabled", async () => {
|
||||
const result = await generateObjectSafe({
|
||||
model: fakeModel(["I cannot answer in JSON, sorry!", '{"ids":[]}']),
|
||||
prompt: "p",
|
||||
schema: Schema,
|
||||
retry: true,
|
||||
});
|
||||
expect(result.object).toEqual({ ids: [] });
|
||||
});
|
||||
|
||||
it("throws the original error when salvage and retry both fail", async () => {
|
||||
await expect(
|
||||
generateObjectSafe({
|
||||
model: fakeModel(["not json at all"]),
|
||||
prompt: "p",
|
||||
schema: Schema,
|
||||
retry: true,
|
||||
}),
|
||||
).rejects.toSatisfy((error: unknown) => NoObjectGeneratedError.isInstance(error));
|
||||
});
|
||||
|
||||
it("does not retry when retry is disabled", async () => {
|
||||
await expect(
|
||||
generateObjectSafe({
|
||||
model: fakeModel(["nope", '{"ids":[]}']),
|
||||
prompt: "p",
|
||||
schema: Schema,
|
||||
}),
|
||||
).rejects.toSatisfy((error: unknown) => NoObjectGeneratedError.isInstance(error));
|
||||
});
|
||||
});
|
||||
135
apps/x/packages/core/src/models/structured.ts
Normal file
135
apps/x/packages/core/src/models/structured.ts
Normal file
|
|
@ -0,0 +1,135 @@
|
|||
import {
|
||||
generateObject,
|
||||
NoObjectGeneratedError,
|
||||
type LanguageModel,
|
||||
type LanguageModelUsage,
|
||||
} from "ai";
|
||||
import type { z } from "zod";
|
||||
import { PrefixLogger } from "@x/shared";
|
||||
|
||||
const log = new PrefixLogger("StructuredOutput");
|
||||
|
||||
const NO_JSON = Symbol("no-json");
|
||||
|
||||
export interface GenerateObjectSafeOptions<T> {
|
||||
model: LanguageModel;
|
||||
system?: string;
|
||||
prompt: string;
|
||||
schema: z.ZodType<T>;
|
||||
/**
|
||||
* Retry once with a reinforced JSON-only instruction when the first
|
||||
* attempt produces unparseable output. Local/small models miss strict
|
||||
* schema output far more often than frontier models, so callers that may
|
||||
* run on a local model should enable this.
|
||||
*/
|
||||
retry?: boolean;
|
||||
}
|
||||
|
||||
export interface GenerateObjectSafeResult<T> {
|
||||
object: T;
|
||||
usage?: LanguageModelUsage;
|
||||
}
|
||||
|
||||
/**
|
||||
* generateObject with degradation paths for models that can't reliably emit
|
||||
* strict JSON: (1) salvage a schema-valid JSON value out of the raw response
|
||||
* text (small models wrap JSON in prose, fences, or <think> blocks), then
|
||||
* (2) optionally retry once with a reinforced instruction. Throws the
|
||||
* original error when nothing works, so callers' failure handling is
|
||||
* unchanged.
|
||||
*/
|
||||
export async function generateObjectSafe<T>(
|
||||
options: GenerateObjectSafeOptions<T>,
|
||||
): Promise<GenerateObjectSafeResult<T>> {
|
||||
try {
|
||||
const result = await generateObject({
|
||||
model: options.model,
|
||||
...(options.system ? { system: options.system } : {}),
|
||||
prompt: options.prompt,
|
||||
schema: options.schema,
|
||||
});
|
||||
return { object: result.object, usage: result.usage };
|
||||
} catch (error) {
|
||||
const salvaged = salvage(error, options.schema);
|
||||
if (salvaged) {
|
||||
log.log("salvaged schema-valid JSON from a malformed response");
|
||||
return salvaged;
|
||||
}
|
||||
if (!options.retry) {
|
||||
throw error;
|
||||
}
|
||||
log.log(
|
||||
`first attempt failed (${error instanceof Error ? error.message : String(error)}); retrying with reinforced JSON instruction`,
|
||||
);
|
||||
try {
|
||||
const system = [
|
||||
options.system ?? "",
|
||||
"Return ONLY a single valid JSON value that matches the requested schema. No prose, no markdown fences, no explanations.",
|
||||
].join("\n\n").trim();
|
||||
const result = await generateObject({
|
||||
model: options.model,
|
||||
system,
|
||||
prompt: options.prompt,
|
||||
schema: options.schema,
|
||||
});
|
||||
return { object: result.object, usage: result.usage };
|
||||
} catch (retryError) {
|
||||
const retrySalvaged = salvage(retryError, options.schema);
|
||||
if (retrySalvaged) {
|
||||
log.log("salvaged schema-valid JSON from the retry response");
|
||||
return retrySalvaged;
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function salvage<T>(
|
||||
error: unknown,
|
||||
schema: z.ZodType<T>,
|
||||
): GenerateObjectSafeResult<T> | null {
|
||||
if (!NoObjectGeneratedError.isInstance(error) || typeof error.text !== "string") {
|
||||
return null;
|
||||
}
|
||||
const candidate = extractJson(error.text);
|
||||
if (candidate === NO_JSON) {
|
||||
return null;
|
||||
}
|
||||
const parsed = schema.safeParse(candidate);
|
||||
if (!parsed.success) {
|
||||
return null;
|
||||
}
|
||||
return { object: parsed.data, usage: error.usage };
|
||||
}
|
||||
|
||||
// Pull a JSON value out of chatty model output: drop <think> blocks, prefer
|
||||
// fenced content, then fall back to the widest parseable {...}/[...] span.
|
||||
function extractJson(raw: string): unknown {
|
||||
let text = raw.replace(/<think>[\s\S]*?<\/think>/gi, "").trim();
|
||||
const fence = text.match(/```(?:json)?\s*([\s\S]*?)```/i);
|
||||
if (fence) {
|
||||
text = fence[1].trim();
|
||||
}
|
||||
try {
|
||||
return JSON.parse(text);
|
||||
} catch {
|
||||
// fall through to span scan
|
||||
}
|
||||
const starts = [text.indexOf("{"), text.indexOf("[")].filter((i) => i >= 0);
|
||||
if (starts.length === 0) {
|
||||
return NO_JSON;
|
||||
}
|
||||
const start = Math.min(...starts);
|
||||
for (let end = text.length; end > start; end--) {
|
||||
const tail = text[end - 1];
|
||||
if (tail !== "}" && tail !== "]") {
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
return JSON.parse(text.slice(start, end));
|
||||
} catch {
|
||||
// keep shrinking
|
||||
}
|
||||
}
|
||||
return NO_JSON;
|
||||
}
|
||||
|
|
@ -1,11 +1,12 @@
|
|||
import { generateObject, type ModelMessage } from "ai";
|
||||
import type { ModelMessage } from "ai";
|
||||
import z from "zod";
|
||||
import { ToolPermissionMetadata } from "@x/shared/dist/runs.js";
|
||||
import { ToolCallPart } from "@x/shared/dist/message.js";
|
||||
import { captureLlmUsage } from "../analytics/usage.js";
|
||||
import { withUseCase, type UseCase } from "../analytics/use_case.js";
|
||||
import { getAutoPermissionDecisionModel, getDefaultModelAndProvider, resolveProviderConfig } from "../models/defaults.js";
|
||||
import { createProvider } from "../models/models.js";
|
||||
import { createLanguageModel } from "../models/models.js";
|
||||
import { generateObjectSafe } from "../models/structured.js";
|
||||
|
||||
const DecisionSchema = z.object({
|
||||
decisions: z.array(z.object({
|
||||
|
|
@ -83,7 +84,7 @@ export async function classifyToolPermissions(input: {
|
|||
const modelId = await getAutoPermissionDecisionModel();
|
||||
const { provider: providerName } = await getDefaultModelAndProvider();
|
||||
const providerConfig = await resolveProviderConfig(providerName);
|
||||
const model = createProvider(providerConfig).languageModel(modelId);
|
||||
const model = createLanguageModel(providerConfig, modelId, { priority: "classifier" });
|
||||
|
||||
const result = await withUseCase(
|
||||
{
|
||||
|
|
@ -91,11 +92,12 @@ export async function classifyToolPermissions(input: {
|
|||
subUseCase: "auto_permission_classifier",
|
||||
...(input.agentName ? { agentName: input.agentName } : {}),
|
||||
},
|
||||
() => generateObject({
|
||||
() => generateObjectSafe({
|
||||
model,
|
||||
system: SYSTEM_PROMPT,
|
||||
prompt: buildPrompt(input),
|
||||
schema: DecisionSchema,
|
||||
retry: true,
|
||||
}),
|
||||
);
|
||||
|
||||
|
|
|
|||
|
|
@ -14,6 +14,11 @@ import type { JsonValue, ModelDescriptor, TurnUsage } from "@x/shared/dist/turns
|
|||
import { convertFromMessages } from "../../agents/runtime.js";
|
||||
import { resolveProviderConfig } from "../../models/defaults.js";
|
||||
import { createProvider } from "../../models/models.js";
|
||||
import {
|
||||
applyLocalModelSettings,
|
||||
isLocalProvider,
|
||||
localLlmScheduler,
|
||||
} from "../../models/local.js";
|
||||
import type {
|
||||
IModelRegistry,
|
||||
LlmStreamEvent,
|
||||
|
|
@ -29,6 +34,10 @@ export type StreamTextInvoker = (options: {
|
|||
messages: ModelMessage[];
|
||||
tools: ToolSet;
|
||||
abortSignal: AbortSignal;
|
||||
temperature?: number;
|
||||
topP?: number;
|
||||
maxOutputTokens?: number;
|
||||
providerOptions?: Record<string, Record<string, JsonValue>>;
|
||||
}) => { fullStream: AsyncIterable<unknown> };
|
||||
|
||||
const defaultInvoker: StreamTextInvoker = (options) =>
|
||||
|
|
@ -61,7 +70,15 @@ export class RealModelRegistry implements IModelRegistry {
|
|||
): Promise<ResolvedModel> {
|
||||
const providerConfig = await this.resolveProvider(descriptor.provider);
|
||||
const provider = this.createProviderImpl(providerConfig);
|
||||
const model = provider.languageModel(descriptor.model);
|
||||
// Local settings (Ollama context window) are applied here, but
|
||||
// scheduling happens per-step in run() where the turn's priority is
|
||||
// known — so priority stays null.
|
||||
const model = applyLocalModelSettings(
|
||||
provider.languageModel(descriptor.model),
|
||||
providerConfig,
|
||||
null,
|
||||
);
|
||||
const local = isLocalProvider(providerConfig);
|
||||
return {
|
||||
descriptor,
|
||||
// The structural -> wire conversion the app uses today: weaves
|
||||
|
|
@ -70,13 +87,14 @@ export class RealModelRegistry implements IModelRegistry {
|
|||
// per-message, so composed requests are byte-stable.
|
||||
encodeMessages: (messages) =>
|
||||
convertFromMessages(messages) as unknown as JsonValue[],
|
||||
stream: (request) => this.run(model, request),
|
||||
stream: (request) => this.run(model, request, local),
|
||||
};
|
||||
}
|
||||
|
||||
private async *run(
|
||||
model: LanguageModel,
|
||||
request: ModelStreamRequest,
|
||||
local: boolean,
|
||||
): AsyncGenerator<LlmStreamEvent, void, void> {
|
||||
const tools: ToolSet = {};
|
||||
for (const descriptor of request.tools) {
|
||||
|
|
@ -93,13 +111,26 @@ export class RealModelRegistry implements IModelRegistry {
|
|||
});
|
||||
}
|
||||
|
||||
const result = this.invoke({
|
||||
model,
|
||||
system: request.systemPrompt,
|
||||
messages: request.messages as ModelMessage[],
|
||||
tools,
|
||||
abortSignal: request.signal,
|
||||
});
|
||||
// Persisted per-call parameters (turn-runtime-design.md §8.3): only
|
||||
// the whitelisted generation knobs are forwarded to the provider.
|
||||
const params = request.parameters ?? {};
|
||||
const generationParams = {
|
||||
...(typeof params.temperature === "number" ? { temperature: params.temperature } : {}),
|
||||
...(typeof params.topP === "number" ? { topP: params.topP } : {}),
|
||||
...(typeof params.maxOutputTokens === "number" ? { maxOutputTokens: params.maxOutputTokens } : {}),
|
||||
...(params.providerOptions && typeof params.providerOptions === "object" && !Array.isArray(params.providerOptions)
|
||||
? { providerOptions: params.providerOptions as Record<string, Record<string, JsonValue>> }
|
||||
: {}),
|
||||
};
|
||||
|
||||
// One scheduler slot per model step: local runtimes serve requests
|
||||
// serially, and headless turns must not starve interactive chat.
|
||||
const release = local
|
||||
? await localLlmScheduler.acquire(
|
||||
request.priority ?? "interactive",
|
||||
request.signal,
|
||||
)
|
||||
: null;
|
||||
|
||||
const parts: Array<z.infer<typeof AssistantContentPart>> = [];
|
||||
let textBuffer = "";
|
||||
|
|
@ -108,99 +139,112 @@ export class RealModelRegistry implements IModelRegistry {
|
|||
let usage: z.infer<typeof TurnUsage> = {};
|
||||
let providerMetadata: JsonValue | undefined;
|
||||
|
||||
for await (const raw of result.fullStream) {
|
||||
request.signal.throwIfAborted();
|
||||
const event = raw as {
|
||||
type: string;
|
||||
text?: string;
|
||||
toolCallId?: string;
|
||||
toolName?: string;
|
||||
input?: unknown;
|
||||
finishReason?: string;
|
||||
usage?: Record<string, number | undefined>;
|
||||
providerMetadata?: unknown;
|
||||
error?: unknown;
|
||||
};
|
||||
switch (event.type) {
|
||||
case "text-start":
|
||||
textBuffer = "";
|
||||
yield { type: "step_event", event: { type: "text_start" } };
|
||||
break;
|
||||
case "text-delta": {
|
||||
const delta = event.text ?? "";
|
||||
textBuffer += delta;
|
||||
const last = parts[parts.length - 1];
|
||||
if (last?.type === "text") {
|
||||
last.text += delta;
|
||||
} else {
|
||||
parts.push({ type: "text", text: delta });
|
||||
try {
|
||||
const result = this.invoke({
|
||||
model,
|
||||
system: request.systemPrompt,
|
||||
messages: request.messages as ModelMessage[],
|
||||
tools,
|
||||
abortSignal: request.signal,
|
||||
...generationParams,
|
||||
});
|
||||
|
||||
for await (const raw of result.fullStream) {
|
||||
request.signal.throwIfAborted();
|
||||
const event = raw as {
|
||||
type: string;
|
||||
text?: string;
|
||||
toolCallId?: string;
|
||||
toolName?: string;
|
||||
input?: unknown;
|
||||
finishReason?: string;
|
||||
usage?: Record<string, number | undefined>;
|
||||
providerMetadata?: unknown;
|
||||
error?: unknown;
|
||||
};
|
||||
switch (event.type) {
|
||||
case "text-start":
|
||||
textBuffer = "";
|
||||
yield { type: "step_event", event: { type: "text_start" } };
|
||||
break;
|
||||
case "text-delta": {
|
||||
const delta = event.text ?? "";
|
||||
textBuffer += delta;
|
||||
const last = parts[parts.length - 1];
|
||||
if (last?.type === "text") {
|
||||
last.text += delta;
|
||||
} else {
|
||||
parts.push({ type: "text", text: delta });
|
||||
}
|
||||
yield { type: "text_delta", delta };
|
||||
break;
|
||||
}
|
||||
yield { type: "text_delta", delta };
|
||||
break;
|
||||
}
|
||||
case "text-end":
|
||||
yield {
|
||||
type: "step_event",
|
||||
event: { type: "text_end", text: textBuffer },
|
||||
};
|
||||
break;
|
||||
case "reasoning-start":
|
||||
reasoningBuffer = "";
|
||||
yield { type: "step_event", event: { type: "reasoning_start" } };
|
||||
break;
|
||||
case "reasoning-delta": {
|
||||
const delta = event.text ?? "";
|
||||
reasoningBuffer += delta;
|
||||
const last = parts[parts.length - 1];
|
||||
if (last?.type === "reasoning") {
|
||||
last.text += delta;
|
||||
} else {
|
||||
parts.push({ type: "reasoning", text: delta });
|
||||
case "text-end":
|
||||
yield {
|
||||
type: "step_event",
|
||||
event: { type: "text_end", text: textBuffer },
|
||||
};
|
||||
break;
|
||||
case "reasoning-start":
|
||||
reasoningBuffer = "";
|
||||
yield { type: "step_event", event: { type: "reasoning_start" } };
|
||||
break;
|
||||
case "reasoning-delta": {
|
||||
const delta = event.text ?? "";
|
||||
reasoningBuffer += delta;
|
||||
const last = parts[parts.length - 1];
|
||||
if (last?.type === "reasoning") {
|
||||
last.text += delta;
|
||||
} else {
|
||||
parts.push({ type: "reasoning", text: delta });
|
||||
}
|
||||
yield { type: "reasoning_delta", delta };
|
||||
break;
|
||||
}
|
||||
yield { type: "reasoning_delta", delta };
|
||||
break;
|
||||
case "reasoning-end":
|
||||
yield {
|
||||
type: "step_event",
|
||||
event: { type: "reasoning_end", text: reasoningBuffer },
|
||||
};
|
||||
break;
|
||||
case "tool-call": {
|
||||
const toolCall = {
|
||||
type: "tool-call" as const,
|
||||
toolCallId: String(event.toolCallId),
|
||||
toolName: String(event.toolName),
|
||||
arguments: event.input,
|
||||
};
|
||||
parts.push(toolCall);
|
||||
yield { type: "step_event", event: { type: "tool_call", toolCall } };
|
||||
break;
|
||||
}
|
||||
case "finish-step": {
|
||||
finishReason = event.finishReason ?? "unknown";
|
||||
usage = mapUsage(event.usage);
|
||||
providerMetadata = toJsonValue(event.providerMetadata);
|
||||
yield {
|
||||
type: "step_event",
|
||||
event: {
|
||||
type: "finish_step",
|
||||
finishReason,
|
||||
usage,
|
||||
...(providerMetadata === undefined
|
||||
? {}
|
||||
: { providerMetadata }),
|
||||
},
|
||||
};
|
||||
break;
|
||||
}
|
||||
case "error":
|
||||
throw event.error instanceof Error
|
||||
? event.error
|
||||
: new Error(formatStreamError(event.error));
|
||||
default:
|
||||
break;
|
||||
}
|
||||
case "reasoning-end":
|
||||
yield {
|
||||
type: "step_event",
|
||||
event: { type: "reasoning_end", text: reasoningBuffer },
|
||||
};
|
||||
break;
|
||||
case "tool-call": {
|
||||
const toolCall = {
|
||||
type: "tool-call" as const,
|
||||
toolCallId: String(event.toolCallId),
|
||||
toolName: String(event.toolName),
|
||||
arguments: event.input,
|
||||
};
|
||||
parts.push(toolCall);
|
||||
yield { type: "step_event", event: { type: "tool_call", toolCall } };
|
||||
break;
|
||||
}
|
||||
case "finish-step": {
|
||||
finishReason = event.finishReason ?? "unknown";
|
||||
usage = mapUsage(event.usage);
|
||||
providerMetadata = toJsonValue(event.providerMetadata);
|
||||
yield {
|
||||
type: "step_event",
|
||||
event: {
|
||||
type: "finish_step",
|
||||
finishReason,
|
||||
usage,
|
||||
...(providerMetadata === undefined
|
||||
? {}
|
||||
: { providerMetadata }),
|
||||
},
|
||||
};
|
||||
break;
|
||||
}
|
||||
case "error":
|
||||
throw event.error instanceof Error
|
||||
? event.error
|
||||
: new Error(formatStreamError(event.error));
|
||||
default:
|
||||
break;
|
||||
}
|
||||
} finally {
|
||||
release?.();
|
||||
}
|
||||
|
||||
yield {
|
||||
|
|
|
|||
|
|
@ -34,6 +34,10 @@ export interface ModelStreamRequest {
|
|||
tools: Array<z.infer<typeof ToolDescriptor>>;
|
||||
parameters: Record<string, JsonValue>;
|
||||
signal: AbortSignal;
|
||||
// Scheduling class for local-model queueing: interactive turns (a human
|
||||
// is watching) preempt queued headless/background work. Defaults to
|
||||
// "interactive" so an unset priority never slows a user down.
|
||||
priority?: "interactive" | "background";
|
||||
}
|
||||
|
||||
export interface ResolvedModel {
|
||||
|
|
|
|||
|
|
@ -951,6 +951,11 @@ class TurnAdvance {
|
|||
tools: composed.tools,
|
||||
parameters: composed.parameters,
|
||||
signal: this.signal,
|
||||
// Headless turns (no human waiting) yield the local-model
|
||||
// queue to interactive chat.
|
||||
priority: this.definition.config.humanAvailable
|
||||
? "interactive"
|
||||
: "background",
|
||||
})) {
|
||||
switch (event.type) {
|
||||
case "text_delta":
|
||||
|
|
|
|||
|
|
@ -566,6 +566,13 @@ const ipcSchemas = {
|
|||
res: z.object({
|
||||
success: z.boolean(),
|
||||
error: z.string().optional(),
|
||||
// Capability caveats from the local-model probe (tool support, context
|
||||
// window) — the connection still succeeded.
|
||||
warnings: z.array(z.string()).optional(),
|
||||
capabilities: z.object({
|
||||
supportsTools: z.boolean().optional(),
|
||||
maxContextLength: z.number().optional(),
|
||||
}).optional(),
|
||||
}),
|
||||
},
|
||||
'models:listForProvider': {
|
||||
|
|
|
|||
|
|
@ -5,6 +5,10 @@ export const LlmProvider = z.object({
|
|||
apiKey: z.string().optional(),
|
||||
baseURL: z.string().optional(),
|
||||
headers: z.record(z.string(), z.string()).optional(),
|
||||
// Context window (in tokens) to request from local runtimes. Ollama defaults
|
||||
// to a ~4k window that silently truncates Rowboat's prompts; when unset,
|
||||
// local providers get a larger default (see core/models/local.ts).
|
||||
contextLength: z.number().int().positive().optional(),
|
||||
});
|
||||
|
||||
export const LlmModelConfig = z.object({
|
||||
|
|
@ -15,6 +19,7 @@ export const LlmModelConfig = z.object({
|
|||
apiKey: z.string().optional(),
|
||||
baseURL: z.string().optional(),
|
||||
headers: z.record(z.string(), z.string()).optional(),
|
||||
contextLength: z.number().int().positive().optional(),
|
||||
model: z.string().optional(),
|
||||
models: z.array(z.string()).optional(),
|
||||
knowledgeGraphModel: z.string().optional(),
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue