feat(llm): add LLM call support to worker IPC chain
- ProviderManager: align API with contracts ProviderAdapter - WorkerProtocol: add llm.request/llm.response message types - WorkerRuntime: add call_llm() for worker→parent→LLM flow - WorkerManager: support tool_registry and provider_manager injection P1-1 complete, P1-2 protocol layer complete. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -7,13 +7,11 @@
|
|||||||
* @module packages/llm/src/ProviderManager
|
* @module packages/llm/src/ProviderManager
|
||||||
*/
|
*/
|
||||||
|
|
||||||
import type { ProviderAdapter } from '@aircoding/contracts'
|
import type { ProviderAdapter, ProviderCompletionInput, ProviderStreamEvent, ProviderCapabilityMatrix, ModelID, ProviderID } from '@aircoding/contracts'
|
||||||
|
|
||||||
// Local type definitions (contract types not yet finalized)
|
|
||||||
type CompleteOptions = { max_tokens?: number; temperature?: number; top_p?: number; system?: string }
|
|
||||||
type StreamEvent = { type: 'text' | 'thinking' | 'done'; content?: string; reason?: string }
|
type StreamEvent = { type: 'text' | 'thinking' | 'done'; content?: string; reason?: string }
|
||||||
type ModelRequirement = { model: string; provider?: string; min_output_tokens?: number; prefers_thinking?: boolean; requires_tools?: boolean }
|
type ModelRequirement = { model: string; provider?: string; min_output_tokens?: number; prefers_thinking?: boolean; requires_tools?: boolean }
|
||||||
type ModelAssignment = { provider: string; model: string; adapter: any; capabilities?: any }
|
type ModelAssignment = { provider: string; model: string; adapter: ProviderAdapter; capabilities?: ReturnType<CapabilityMatrixRegistry['lookup']> }
|
||||||
|
|
||||||
import { ModelConfigLoader, createModelConfigLoader } from './ModelConfigLoader.js'
|
import { ModelConfigLoader, createModelConfigLoader } from './ModelConfigLoader.js'
|
||||||
import { CapabilityMatrixRegistry, createCapabilityMatrixRegistry } from './CapabilityMatrix.js'
|
import { CapabilityMatrixRegistry, createCapabilityMatrixRegistry } from './CapabilityMatrix.js'
|
||||||
@@ -80,34 +78,118 @@ export class ProviderManager {
|
|||||||
|
|
||||||
/**
|
/**
|
||||||
* Complete a request with the current model.
|
* Complete a request with the current model.
|
||||||
|
* Returns AsyncIterable of ProviderStreamEvent per contracts §15.
|
||||||
*/
|
*/
|
||||||
async complete(
|
async *complete(
|
||||||
|
input: ProviderCompletionInput
|
||||||
|
): AsyncGenerator<ProviderStreamEvent> {
|
||||||
|
const adapter = this.current_adapter || this.adapters.get(input.provider_id || 'anthropic')
|
||||||
|
if (!adapter) {
|
||||||
|
throw new Error('No adapter selected. Call select_model first.')
|
||||||
|
}
|
||||||
|
|
||||||
|
yield* adapter.complete(input)
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Complete and collect all events into a single response.
|
||||||
|
*/
|
||||||
|
async complete_text(
|
||||||
|
messages: unknown[],
|
||||||
|
options: { model?: string; max_tokens?: number; temperature?: number; system?: string } = {}
|
||||||
|
): Promise<{ content: string; usage?: { input_tokens: number; output_tokens: number } }> {
|
||||||
|
const adapter = this.current_adapter
|
||||||
|
if (!adapter) {
|
||||||
|
throw new Error('No adapter selected. Call select_model first.')
|
||||||
|
}
|
||||||
|
|
||||||
|
const model_id = options.model || 'claude-haiku-4-5-20251001'
|
||||||
|
const input: ProviderCompletionInput = {
|
||||||
|
provider_id: 'anthropic',
|
||||||
|
model_id: model_id as ModelID,
|
||||||
|
canonical_format: 'anthropic',
|
||||||
|
messages,
|
||||||
|
max_output_tokens: options.max_tokens || 4096,
|
||||||
|
temperature: options.temperature,
|
||||||
|
system: options.system
|
||||||
|
}
|
||||||
|
|
||||||
|
let content = ''
|
||||||
|
let usage: { input_tokens: number; output_tokens: number } | undefined
|
||||||
|
|
||||||
|
for await (const event of adapter.complete(input)) {
|
||||||
|
if (event.type === 'content_delta') {
|
||||||
|
const payload = event.payload as { type: string; text?: string; thinking?: string }
|
||||||
|
if (payload.type === 'text_delta') {
|
||||||
|
content += payload.text || ''
|
||||||
|
} else if (payload.type === 'thinking_delta') {
|
||||||
|
// Accumulate thinking for reference but don't include in content
|
||||||
|
}
|
||||||
|
} else if (event.type === 'message_stop') {
|
||||||
|
const payload = event.payload as { usage?: { output_tokens: number } }
|
||||||
|
if (payload.usage) {
|
||||||
|
usage = { input_tokens: 0, output_tokens: payload.usage.output_tokens }
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return { content, usage }
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Stream a completion request (passthrough to adapter).
|
||||||
|
*/
|
||||||
|
async *stream_complete(
|
||||||
|
input: ProviderCompletionInput
|
||||||
|
): AsyncGenerator<ProviderStreamEvent> {
|
||||||
|
const adapter = this.current_adapter
|
||||||
|
if (!adapter) {
|
||||||
|
throw new Error('No adapter selected. Call select_model first.')
|
||||||
|
}
|
||||||
|
|
||||||
|
yield* adapter.complete(input)
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Legacy overload: select model then complete.
|
||||||
|
*/
|
||||||
|
async complete_after_select(
|
||||||
messages: unknown[],
|
messages: unknown[],
|
||||||
assignment: ModelAssignment,
|
assignment: ModelAssignment,
|
||||||
options: CompleteOptions = {}
|
options: { max_tokens?: number; temperature?: number } = {}
|
||||||
): Promise<{ content: string; usage?: { input_tokens: number; output_tokens: number } }> {
|
): Promise<{ content: string; usage?: { input_tokens: number; output_tokens: number } }> {
|
||||||
const adapter = assignment.adapter || this.current_adapter
|
const adapter = assignment.adapter || this.current_adapter
|
||||||
if (!adapter) {
|
if (!adapter) {
|
||||||
throw new Error('No adapter selected. Call select_model first.')
|
throw new Error('No adapter selected. Call select_model first.')
|
||||||
}
|
}
|
||||||
|
|
||||||
return adapter.complete(messages as any, { model: assignment.model }, options)
|
const input: ProviderCompletionInput = {
|
||||||
|
provider_id: assignment.provider as ProviderID || 'anthropic',
|
||||||
|
model_id: assignment.model as ModelID,
|
||||||
|
canonical_format: 'anthropic',
|
||||||
|
messages,
|
||||||
|
max_output_tokens: options.max_tokens || 4096,
|
||||||
|
temperature: options.temperature
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
let content = ''
|
||||||
* Stream a completion request.
|
let usage: { input_tokens: number; output_tokens: number } | undefined
|
||||||
*/
|
|
||||||
async *stream_complete(
|
for await (const event of adapter.complete(input)) {
|
||||||
messages: unknown[],
|
if (event.type === 'content_delta') {
|
||||||
assignment: ModelAssignment,
|
const payload = event.payload as { type: string; text?: string }
|
||||||
options: CompleteOptions = {}
|
if (payload.type === 'text_delta') {
|
||||||
): AsyncGenerator<StreamEvent> {
|
content += payload.text || ''
|
||||||
const adapter = assignment.adapter || this.current_adapter
|
}
|
||||||
if (!adapter) {
|
} else if (event.type === 'message_stop') {
|
||||||
throw new Error('No adapter selected. Call select_model first.')
|
const payload = event.payload as { usage?: { output_tokens: number } }
|
||||||
|
if (payload.usage) {
|
||||||
|
usage = { input_tokens: 0, output_tokens: payload.usage.output_tokens }
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
yield* adapter.stream_complete(messages as any, { model: assignment.model }, options)
|
return { content, usage }
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|||||||
@@ -13,6 +13,8 @@ import type { ChildProcess } from 'child_process'
|
|||||||
import { WorkerProcess, type WorkerExitCode } from './WorkerProcess.js'
|
import { WorkerProcess, type WorkerExitCode } from './WorkerProcess.js'
|
||||||
import { WorkerProtocol, type WorkerMessage, type WorkerMessageType } from './WorkerProtocol.js'
|
import { WorkerProtocol, type WorkerMessage, type WorkerMessageType } from './WorkerProtocol.js'
|
||||||
import type { WorkerResult, WorkerStatus, AgentType } from '@aircoding/contracts'
|
import type { WorkerResult, WorkerStatus, AgentType } from '@aircoding/contracts'
|
||||||
|
import type { ToolRegistry } from '../tools/ToolRegistry.js'
|
||||||
|
import type { ProviderManager } from '@aircoding/llm'
|
||||||
|
|
||||||
export interface WorkerConfig {
|
export interface WorkerConfig {
|
||||||
entrypoint: string // Path to worker main.ts
|
entrypoint: string // Path to worker main.ts
|
||||||
@@ -36,9 +38,36 @@ export interface WorkerHandle {
|
|||||||
export class WorkerManager {
|
export class WorkerManager {
|
||||||
private protocol: WorkerProtocol
|
private protocol: WorkerProtocol
|
||||||
private workers: Map<string, WorkerHandle> = new Map()
|
private workers: Map<string, WorkerHandle> = new Map()
|
||||||
|
private tool_registry?: ToolRegistry
|
||||||
|
private provider_manager?: ProviderManager
|
||||||
|
private execution_context?: { session_id: string; project_id: string; project_root: string }
|
||||||
|
|
||||||
constructor() {
|
constructor(tool_registry?: ToolRegistry, provider_manager?: ProviderManager) {
|
||||||
this.protocol = new WorkerProtocol()
|
this.protocol = new WorkerProtocol()
|
||||||
|
this.tool_registry = tool_registry
|
||||||
|
this.provider_manager = provider_manager
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Set execution context (session_id, project_id, project_root).
|
||||||
|
* Required for tool/LLM calls from workers.
|
||||||
|
*/
|
||||||
|
set_context(ctx: { session_id: string; project_id: string; project_root: string }): void {
|
||||||
|
this.execution_context = ctx
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Set tool registry (can be injected after construction).
|
||||||
|
*/
|
||||||
|
set_tool_registry(registry: ToolRegistry): void {
|
||||||
|
this.tool_registry = registry
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Set provider manager (can be injected after construction).
|
||||||
|
*/
|
||||||
|
set_provider_manager(pm: ProviderManager): void {
|
||||||
|
this.provider_manager = pm
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|||||||
@@ -27,6 +27,7 @@ export type WorkerMessageType =
|
|||||||
| 'tool.result'
|
| 'tool.result'
|
||||||
| 'agent.cancel'
|
| 'agent.cancel'
|
||||||
| 'agent.ping'
|
| 'agent.ping'
|
||||||
|
| 'llm.response'
|
||||||
// Worker → Parent
|
// Worker → Parent
|
||||||
| 'worker.ready'
|
| 'worker.ready'
|
||||||
| 'tool.call'
|
| 'tool.call'
|
||||||
@@ -35,6 +36,7 @@ export type WorkerMessageType =
|
|||||||
| 'worker.heartbeat'
|
| 'worker.heartbeat'
|
||||||
| 'worker.error'
|
| 'worker.error'
|
||||||
| 'event'
|
| 'event'
|
||||||
|
| 'llm.request'
|
||||||
|
|
||||||
const PROTOCOL_VERSION = 1
|
const PROTOCOL_VERSION = 1
|
||||||
|
|
||||||
@@ -44,13 +46,15 @@ const DIRECTION_RULES: Record<string, WorkerMessageDirection> = {
|
|||||||
'tool.result': 'parent_to_worker',
|
'tool.result': 'parent_to_worker',
|
||||||
'agent.cancel': 'parent_to_worker',
|
'agent.cancel': 'parent_to_worker',
|
||||||
'agent.ping': 'parent_to_worker',
|
'agent.ping': 'parent_to_worker',
|
||||||
|
'llm.response': 'parent_to_worker',
|
||||||
'worker.ready': 'worker_to_parent',
|
'worker.ready': 'worker_to_parent',
|
||||||
'tool.call': 'worker_to_parent',
|
'tool.call': 'worker_to_parent',
|
||||||
'worker.result': 'worker_to_parent',
|
'worker.result': 'worker_to_parent',
|
||||||
'worker.checkpoint': 'worker_to_parent',
|
'worker.checkpoint': 'worker_to_parent',
|
||||||
'worker.heartbeat': 'worker_to_parent',
|
'worker.heartbeat': 'worker_to_parent',
|
||||||
'worker.error': 'worker_to_parent',
|
'worker.error': 'worker_to_parent',
|
||||||
'event': 'worker_to_parent'
|
'event': 'worker_to_parent',
|
||||||
|
'llm.request': 'worker_to_parent'
|
||||||
}
|
}
|
||||||
|
|
||||||
export class WorkerProtocol {
|
export class WorkerProtocol {
|
||||||
|
|||||||
@@ -24,6 +24,20 @@ export interface ToolCallResult {
|
|||||||
content: Record<string, unknown>
|
content: Record<string, unknown>
|
||||||
}
|
}
|
||||||
|
|
||||||
|
export interface LLMRequest {
|
||||||
|
messages: Array<{ role: string; content: unknown }>
|
||||||
|
model?: string
|
||||||
|
max_tokens?: number
|
||||||
|
temperature?: number
|
||||||
|
tools?: unknown[]
|
||||||
|
}
|
||||||
|
|
||||||
|
export interface LLMResponse {
|
||||||
|
content: string
|
||||||
|
usage?: { input_tokens: number; output_tokens: number }
|
||||||
|
tool_calls?: Array<{ name: string; arguments: Record<string, unknown> }>
|
||||||
|
}
|
||||||
|
|
||||||
export class WorkerRuntime {
|
export class WorkerRuntime {
|
||||||
private agent_id: string
|
private agent_id: string
|
||||||
private session_id: string
|
private session_id: string
|
||||||
@@ -63,6 +77,36 @@ export class WorkerRuntime {
|
|||||||
return promise
|
return promise
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Call LLM through parent process via IPC.
|
||||||
|
* INV-3: This is the ONLY way workers access LLM.
|
||||||
|
*/
|
||||||
|
async call_llm(request: LLMRequest): Promise<LLMResponse> {
|
||||||
|
const call_id = crypto.randomUUID()
|
||||||
|
|
||||||
|
const promise = new Promise<LLMResponse>((resolve, reject) => {
|
||||||
|
this.pending_calls.set(call_id, { resolve: resolve as any, reject })
|
||||||
|
|
||||||
|
// Set timeout (LLM calls can be long)
|
||||||
|
setTimeout(() => {
|
||||||
|
this.pending_calls.delete(call_id)
|
||||||
|
reject(new Error(`LLM call timeout: ${request.model || 'default'}`))
|
||||||
|
}, 300000) // 5 minutes
|
||||||
|
})
|
||||||
|
|
||||||
|
// Send llm.request via IPC
|
||||||
|
this.send_message('llm.request', {
|
||||||
|
call_id,
|
||||||
|
messages: request.messages,
|
||||||
|
model: request.model || 'claude-haiku-4-5-20251001',
|
||||||
|
max_tokens: request.max_tokens || 4096,
|
||||||
|
temperature: request.temperature,
|
||||||
|
tools: request.tools
|
||||||
|
})
|
||||||
|
|
||||||
|
return promise
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Emit an event to the parent.
|
* Emit an event to the parent.
|
||||||
*/
|
*/
|
||||||
@@ -118,6 +162,22 @@ export class WorkerRuntime {
|
|||||||
// Respond to ping
|
// Respond to ping
|
||||||
this.send_message('worker.heartbeat', { timestamp: new Date().toISOString() })
|
this.send_message('worker.heartbeat', { timestamp: new Date().toISOString() })
|
||||||
break
|
break
|
||||||
|
|
||||||
|
case 'llm.response': {
|
||||||
|
const call_id = payload.call_id as string
|
||||||
|
const pending = this.pending_calls.get(call_id)
|
||||||
|
if (pending) {
|
||||||
|
this.pending_calls.delete(call_id)
|
||||||
|
// Resolve with LLM response
|
||||||
|
const response: LLMResponse = {
|
||||||
|
content: (payload.content as string) || '',
|
||||||
|
usage: payload.usage as LLMResponse['usage'],
|
||||||
|
tool_calls: payload.tool_calls as LLMResponse['tool_calls']
|
||||||
|
}
|
||||||
|
pending.resolve(response as any)
|
||||||
|
}
|
||||||
|
break
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user