File size: 3,984 Bytes
25d91a4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 | // Streams LLM responses through registered providers and normalizes events.
// This facade owns the process-default AI runtime wiring: it installs the
// OpenClaw host policy ports and registers built-in providers exactly once,
// before any caller imports the stream API.
import { defaultApiRegistry, defaultLlmRuntime } from "@openclaw/ai/internal/runtime";
import { registerBuiltInApiProviders } from "@openclaw/ai/providers";
import { classifyGatewayStorageFailure } from "../infra/sqlite-error-diagnostics.js";
import { getModelLlmRuntime } from "./model-runtime-binding.js";
import "./ai-transport-host.js";
import type {
Api,
AssistantMessage,
AssistantMessageEventStreamContract,
Context,
Model,
ProviderStreamOptions,
SimpleStreamOptions,
} from "./types.js";
import { createAssistantMessageEventStream } from "./utils/event-stream.js";
registerBuiltInApiProviders(defaultApiRegistry);
let transportRuntimeHostPromise: Promise<void> | undefined;
async function ensureTransportRuntimeHost(): Promise<void> {
// Async completion entry points install heavy provider ports before the runtime
// can invoke them, without adding their plugin graph to this eager facade.
transportRuntimeHostPromise ??= import("../agents/ai-transport-runtime-host.js").then(
({ configureAiTransportRuntimeHost }) => configureAiTransportRuntimeHost(),
);
await transportRuntimeHostPromise;
}
function createRuntimeHostErrorMessage(model: Model, error: unknown): AssistantMessage {
return {
role: "assistant",
content: [],
api: model.api,
provider: model.provider,
model: model.id,
usage: {
input: 0,
output: 0,
cacheRead: 0,
cacheWrite: 0,
totalTokens: 0,
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
},
stopReason: "error",
errorMessage: error instanceof Error ? error.message : String(error),
errorCode: classifyGatewayStorageFailure(error),
timestamp: Date.now(),
};
}
function deferUntilTransportRuntimeHost(
model: Model,
start: () => AssistantMessageEventStreamContract,
): AssistantMessageEventStreamContract {
const output = createAssistantMessageEventStream();
void (async () => {
try {
await ensureTransportRuntimeHost();
for await (const event of start()) {
output.push(event);
}
} catch (error) {
const message = createRuntimeHostErrorMessage(model, error);
output.push({ type: "error", reason: "error", error: message });
} finally {
output.end();
}
})();
return output;
}
function resolveRuntime(model: Model) {
return getModelLlmRuntime(model) ?? defaultLlmRuntime;
}
export function stream<TApi extends Api>(
model: Model<TApi>,
context: Context,
options?: ProviderStreamOptions,
): AssistantMessageEventStreamContract {
return deferUntilTransportRuntimeHost(model, () =>
resolveRuntime(model).stream(model, context, options),
);
}
export async function complete<TApi extends Api>(
model: Model<TApi>,
context: Context,
options?: ProviderStreamOptions,
): Promise<AssistantMessage> {
await ensureTransportRuntimeHost();
return await resolveRuntime(model).complete(model, context, options);
}
export function streamSimple<TApi extends Api>(
model: Model<TApi>,
context: Context,
options?: SimpleStreamOptions,
): AssistantMessageEventStreamContract {
return deferUntilTransportRuntimeHost(model, () =>
resolveRuntime(model).streamSimple(model, context, options),
);
}
export async function completeSimple<TApi extends Api>(
model: Model<TApi>,
context: Context,
options?: SimpleStreamOptions,
assertCurrent?: () => void,
): Promise<AssistantMessage> {
await ensureTransportRuntimeHost();
// Runtime setup can outlive its caller. Admit only a current request to the provider.
assertCurrent?.();
options?.signal?.throwIfAborted();
return await resolveRuntime(model).completeSimple(model, context, options);
}
|