mirror of
https://github.com/tiennm99/ccs.git
synced 2026-07-16 06:16:37 +00:00
refactor(glmt): modularize transformer pipeline
- extract 7 pipeline modules (1122 → 206 lines) - request-transformer, stream-parser, response-builder - tool-call-handler, content-transformer - shared types.ts for pipeline interfaces Phase 3: Large Files Breakdown (glmt-transformer.ts)
This commit is contained in:
kaitranntt
+96
-1012
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,183 @@
|
||||
/**
|
||||
* ContentTransformer - Handle message sanitization and content transformations
|
||||
*
|
||||
* Responsibilities:
|
||||
* - Sanitize messages for OpenAI API compatibility
|
||||
* - Extract thinking control tags from messages
|
||||
* - Detect think keywords in user prompts
|
||||
* - Transform tools between Anthropic and OpenAI formats
|
||||
*/
|
||||
|
||||
import type { Message, ContentBlock, AnthropicTool, OpenAITool, ThinkingConfig } from './types';
|
||||
|
||||
export class ContentTransformer {
|
||||
private defaultThinking: boolean;
|
||||
|
||||
constructor(defaultThinking = true) {
|
||||
this.defaultThinking = defaultThinking;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sanitize messages for OpenAI API compatibility
|
||||
*/
|
||||
sanitizeMessages(messages: Message[]): Message[] {
|
||||
const result: Message[] = [];
|
||||
|
||||
for (const msg of messages) {
|
||||
// If content is a string, add as-is
|
||||
if (typeof msg.content === 'string') {
|
||||
result.push(msg);
|
||||
continue;
|
||||
}
|
||||
|
||||
// If content is an array, process blocks
|
||||
if (Array.isArray(msg.content)) {
|
||||
// Separate tool_result blocks from other content
|
||||
const toolResults = msg.content.filter((block) => block.type === 'tool_result');
|
||||
const textBlocks = msg.content.filter((block) => block.type === 'text');
|
||||
|
||||
// CRITICAL: Tool messages must come BEFORE user text in OpenAI API
|
||||
for (const toolResult of toolResults) {
|
||||
result.push({
|
||||
role: 'tool',
|
||||
content:
|
||||
typeof toolResult.content === 'string'
|
||||
? toolResult.content
|
||||
: JSON.stringify(toolResult.content),
|
||||
} as Message & { tool_call_id: string });
|
||||
}
|
||||
|
||||
// Add text content as user/assistant message AFTER tool messages
|
||||
if (textBlocks.length > 0) {
|
||||
const textContent =
|
||||
textBlocks.length === 1
|
||||
? textBlocks[0].text || ''
|
||||
: textBlocks.map((b) => b.text || '').join('\n');
|
||||
|
||||
result.push({
|
||||
role: msg.role,
|
||||
content: textContent,
|
||||
});
|
||||
}
|
||||
|
||||
// If no content at all, add empty message
|
||||
if (textBlocks.length === 0 && toolResults.length === 0) {
|
||||
result.push({
|
||||
role: msg.role,
|
||||
content: '',
|
||||
});
|
||||
}
|
||||
|
||||
continue;
|
||||
}
|
||||
|
||||
// Fallback: return message as-is
|
||||
result.push(msg);
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
/**
|
||||
* Transform Anthropic tools to OpenAI tools format
|
||||
*/
|
||||
transformTools(anthropicTools: AnthropicTool[]): OpenAITool[] {
|
||||
return anthropicTools.map((tool) => ({
|
||||
type: 'function' as const,
|
||||
function: {
|
||||
name: tool.name,
|
||||
description: tool.description,
|
||||
parameters: tool.input_schema || {},
|
||||
},
|
||||
}));
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if messages contain thinking control tags
|
||||
*/
|
||||
hasThinkingTags(messages: Message[]): boolean {
|
||||
for (const msg of messages) {
|
||||
if (msg.role !== 'user') continue;
|
||||
const content = msg.content;
|
||||
if (typeof content !== 'string') continue;
|
||||
|
||||
// Check for control tags
|
||||
if (/<Thinking:(On|Off)>/i.test(content) || /<Effort:(Low|Medium|High)>/i.test(content)) {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
/**
|
||||
* Extract thinking control tags from user messages
|
||||
*/
|
||||
extractThinkingControl(messages: Message[]): ThinkingConfig {
|
||||
const config: ThinkingConfig = {
|
||||
thinking: this.defaultThinking,
|
||||
effort: 'medium',
|
||||
};
|
||||
|
||||
// Scan user messages for control tags
|
||||
for (const msg of messages) {
|
||||
if (msg.role !== 'user') continue;
|
||||
|
||||
const content = msg.content;
|
||||
if (typeof content !== 'string') continue;
|
||||
|
||||
// Check for <Thinking:On|Off>
|
||||
const thinkingMatch = content.match(/<Thinking:(On|Off)>/i);
|
||||
if (thinkingMatch) {
|
||||
config.thinking = thinkingMatch[1].toLowerCase() === 'on';
|
||||
}
|
||||
|
||||
// Check for <Effort:Low|Medium|High>
|
||||
const effortMatch = content.match(/<Effort:(Low|Medium|High)>/i);
|
||||
if (effortMatch) {
|
||||
config.effort = effortMatch[1].toLowerCase();
|
||||
}
|
||||
}
|
||||
|
||||
return config;
|
||||
}
|
||||
|
||||
/**
|
||||
* Detect Anthropic-style "think" keywords in user prompts
|
||||
*/
|
||||
detectThinkKeywords(
|
||||
messages: Message[]
|
||||
): { thinking: boolean; effort: string; keyword: string } | null {
|
||||
if (!messages || messages.length === 0) return null;
|
||||
|
||||
// Extract text from user messages
|
||||
const text = messages
|
||||
.filter((m) => m.role === 'user')
|
||||
.map((m) => {
|
||||
if (typeof m.content === 'string') return m.content;
|
||||
if (Array.isArray(m.content)) {
|
||||
return (m.content as ContentBlock[])
|
||||
.filter((block) => block.type === 'text')
|
||||
.map((block) => block.text || '')
|
||||
.join(' ');
|
||||
}
|
||||
return '';
|
||||
})
|
||||
.join(' ');
|
||||
|
||||
// Priority: ultrathink > think harder > think hard > think
|
||||
if (/\bultrathink\b/i.test(text)) {
|
||||
return { thinking: true, effort: 'max', keyword: 'ultrathink' };
|
||||
}
|
||||
if (/\bthink\s+harder\b/i.test(text)) {
|
||||
return { thinking: true, effort: 'high', keyword: 'think harder' };
|
||||
}
|
||||
if (/\bthink\s+hard\b/i.test(text)) {
|
||||
return { thinking: true, effort: 'medium', keyword: 'think hard' };
|
||||
}
|
||||
if (/\bthink\b/i.test(text)) {
|
||||
return { thinking: true, effort: 'low', keyword: 'think' };
|
||||
}
|
||||
|
||||
return null; // No keywords detected
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,35 @@
|
||||
/**
|
||||
* Pipeline Module Exports
|
||||
*
|
||||
* Barrel file for the GLMT transformation pipeline
|
||||
*/
|
||||
|
||||
// Types
|
||||
export type {
|
||||
ContentBlock,
|
||||
Message,
|
||||
AnthropicTool,
|
||||
OpenAITool,
|
||||
OpenAIToolCall,
|
||||
OpenAIToolCallDelta,
|
||||
AnthropicRequest,
|
||||
OpenAIRequest,
|
||||
ThinkingConfig,
|
||||
TransformResult,
|
||||
ThinkingSignature,
|
||||
OpenAIChoice,
|
||||
OpenAIResponse,
|
||||
AnthropicResponse,
|
||||
SSEEvent,
|
||||
AnthropicSSEEvent,
|
||||
AccumulatorBlock,
|
||||
ValidationResult,
|
||||
GlmtTransformerConfig,
|
||||
} from './types';
|
||||
|
||||
// Pipeline components
|
||||
export { ContentTransformer } from './content-transformer';
|
||||
export { ToolCallHandler } from './tool-call-handler';
|
||||
export { ResponseBuilder } from './response-builder';
|
||||
export { StreamParser, type StreamParserConfig } from './stream-parser';
|
||||
export { RequestTransformer, type RequestTransformerConfig } from './request-transformer';
|
||||
@@ -0,0 +1,161 @@
|
||||
/**
|
||||
* RequestTransformer - Handle Anthropic → OpenAI request transformation
|
||||
*
|
||||
* Responsibilities:
|
||||
* - Transform Anthropic request format to OpenAI format
|
||||
* - Inject locale and reasoning instructions
|
||||
* - Map models and configure parameters
|
||||
*/
|
||||
|
||||
import { LocaleEnforcer } from '../locale-enforcer';
|
||||
import { ReasoningEnforcer } from '../reasoning-enforcer';
|
||||
import { ContentTransformer } from './content-transformer';
|
||||
import type {
|
||||
Message,
|
||||
AnthropicRequest,
|
||||
OpenAIRequest,
|
||||
ThinkingConfig,
|
||||
TransformResult,
|
||||
} from './types';
|
||||
|
||||
export interface RequestTransformerConfig {
|
||||
defaultThinking?: boolean;
|
||||
verbose?: boolean;
|
||||
explicitReasoning?: boolean;
|
||||
log?: (message: string) => void;
|
||||
}
|
||||
|
||||
export class RequestTransformer {
|
||||
private localeEnforcer: LocaleEnforcer;
|
||||
private reasoningEnforcer: ReasoningEnforcer;
|
||||
private contentTransformer: ContentTransformer;
|
||||
private log: (message: string) => void;
|
||||
private modelMaxTokens: Record<string, number>;
|
||||
|
||||
constructor(config: RequestTransformerConfig = {}) {
|
||||
const defaultThinking = config.defaultThinking ?? true;
|
||||
this.log = config.log || (() => {});
|
||||
|
||||
this.localeEnforcer = new LocaleEnforcer();
|
||||
this.reasoningEnforcer = new ReasoningEnforcer({
|
||||
enabled: config.explicitReasoning ?? true,
|
||||
});
|
||||
this.contentTransformer = new ContentTransformer(defaultThinking);
|
||||
|
||||
this.modelMaxTokens = {
|
||||
'GLM-4.6': 128000,
|
||||
'GLM-4.5': 96000,
|
||||
'GLM-4.5-air': 16000,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Transform Anthropic request to OpenAI format
|
||||
*/
|
||||
transform(anthropicRequest: AnthropicRequest): TransformResult {
|
||||
try {
|
||||
const messages = anthropicRequest.messages || [];
|
||||
|
||||
// 1. Extract thinking control from messages
|
||||
const thinkingConfig = this.contentTransformer.extractThinkingControl(messages);
|
||||
const hasControlTags = this.contentTransformer.hasThinkingTags(messages);
|
||||
|
||||
// 2. Detect "think" keywords in user prompts
|
||||
const keywordConfig = this.contentTransformer.detectThinkKeywords(messages);
|
||||
if (keywordConfig && !anthropicRequest.thinking && !hasControlTags) {
|
||||
thinkingConfig.thinking = keywordConfig.thinking;
|
||||
thinkingConfig.effort = keywordConfig.effort;
|
||||
this.log(
|
||||
`Detected think keyword: ${keywordConfig.keyword}, effort=${keywordConfig.effort}`
|
||||
);
|
||||
}
|
||||
|
||||
// 3. Check anthropicRequest.thinking parameter (takes precedence)
|
||||
if (anthropicRequest.thinking) {
|
||||
if (anthropicRequest.thinking.type === 'enabled') {
|
||||
thinkingConfig.thinking = true;
|
||||
this.log('Claude CLI explicitly enabled thinking');
|
||||
} else if (anthropicRequest.thinking.type === 'disabled') {
|
||||
thinkingConfig.thinking = false;
|
||||
this.log('Claude CLI explicitly disabled thinking');
|
||||
}
|
||||
}
|
||||
|
||||
this.log(`Final thinking control: ${JSON.stringify(thinkingConfig)}`);
|
||||
|
||||
// 4. Map model
|
||||
const glmModel = this.mapModel(anthropicRequest.model);
|
||||
|
||||
// 5. Inject locale instruction
|
||||
const messagesWithLocale = this.localeEnforcer.injectInstruction(
|
||||
messages as Parameters<typeof this.localeEnforcer.injectInstruction>[0]
|
||||
) as unknown as Message[];
|
||||
|
||||
// 6. Inject reasoning instruction
|
||||
const messagesWithReasoning = this.reasoningEnforcer.injectInstruction(
|
||||
messagesWithLocale as unknown as Parameters<
|
||||
typeof this.reasoningEnforcer.injectInstruction
|
||||
>[0],
|
||||
thinkingConfig
|
||||
) as unknown as Message[];
|
||||
|
||||
// 7. Build OpenAI request
|
||||
const openaiRequest: OpenAIRequest = {
|
||||
model: glmModel,
|
||||
messages: this.contentTransformer.sanitizeMessages(messagesWithReasoning),
|
||||
max_tokens: this.getMaxTokens(glmModel),
|
||||
stream: anthropicRequest.stream ?? false,
|
||||
};
|
||||
|
||||
// 8. Transform tools if present
|
||||
if (anthropicRequest.tools && anthropicRequest.tools.length > 0) {
|
||||
openaiRequest.tools = this.contentTransformer.transformTools(anthropicRequest.tools);
|
||||
openaiRequest.tool_choice = 'auto';
|
||||
this.log(`Transformed ${anthropicRequest.tools.length} tools for OpenAI format`);
|
||||
}
|
||||
|
||||
// 9. Preserve optional parameters
|
||||
if (anthropicRequest.temperature !== undefined) {
|
||||
openaiRequest.temperature = anthropicRequest.temperature;
|
||||
}
|
||||
if (anthropicRequest.top_p !== undefined) {
|
||||
openaiRequest.top_p = anthropicRequest.top_p;
|
||||
}
|
||||
if (anthropicRequest.stream !== undefined) {
|
||||
openaiRequest.stream = anthropicRequest.stream;
|
||||
}
|
||||
|
||||
// 10. Inject reasoning parameters
|
||||
this.injectReasoningParams(openaiRequest, thinkingConfig);
|
||||
|
||||
return { openaiRequest, thinkingConfig };
|
||||
} catch (error) {
|
||||
const err = error as Error;
|
||||
console.error('[RequestTransformer] Transformation error:', err);
|
||||
return {
|
||||
openaiRequest: anthropicRequest,
|
||||
thinkingConfig: { thinking: false, effort: 'medium' },
|
||||
error: err.message,
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
private injectReasoningParams(
|
||||
openaiRequest: OpenAIRequest,
|
||||
thinkingConfig: ThinkingConfig
|
||||
): void {
|
||||
openaiRequest.do_sample = true;
|
||||
if (thinkingConfig.thinking) {
|
||||
openaiRequest.reasoning = true;
|
||||
openaiRequest.reasoning_effort = thinkingConfig.effort;
|
||||
}
|
||||
}
|
||||
|
||||
private mapModel(_anthropicModel: string): string {
|
||||
return 'GLM-4.6';
|
||||
}
|
||||
|
||||
private getMaxTokens(model: string): number {
|
||||
return this.modelMaxTokens[model] || 128000;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,202 @@
|
||||
/**
|
||||
* ResponseBuilder - Create SSE events for Anthropic streaming format
|
||||
*
|
||||
* Responsibilities:
|
||||
* - Create message_start, message_delta, message_stop events
|
||||
* - Create content_block_start, content_block_delta, content_block_stop events
|
||||
* - Generate thinking signatures
|
||||
* - Map stop reasons between formats
|
||||
*/
|
||||
|
||||
import * as crypto from 'crypto';
|
||||
import type { DeltaAccumulator } from '../delta-accumulator';
|
||||
import type { AccumulatorBlock, AnthropicSSEEvent, ThinkingSignature } from './types';
|
||||
|
||||
export class ResponseBuilder {
|
||||
private verbose: boolean;
|
||||
|
||||
constructor(verbose = false) {
|
||||
this.verbose = verbose;
|
||||
}
|
||||
|
||||
/**
|
||||
* Create message_start event
|
||||
*/
|
||||
createMessageStartEvent(accumulator: DeltaAccumulator): AnthropicSSEEvent {
|
||||
return {
|
||||
event: 'message_start',
|
||||
data: {
|
||||
type: 'message_start',
|
||||
message: {
|
||||
id: accumulator.getMessageId(),
|
||||
type: 'message',
|
||||
role: accumulator.getRole(),
|
||||
content: [],
|
||||
model: accumulator.getModel() || 'glm-4.6',
|
||||
stop_reason: null,
|
||||
usage: {
|
||||
input_tokens: accumulator.getInputTokens(),
|
||||
output_tokens: 0,
|
||||
},
|
||||
},
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Create content_block_start event
|
||||
*/
|
||||
createContentBlockStartEvent(block: AccumulatorBlock): AnthropicSSEEvent {
|
||||
return {
|
||||
event: 'content_block_start',
|
||||
data: {
|
||||
type: 'content_block_start',
|
||||
index: block.index,
|
||||
content_block: {
|
||||
type: block.type,
|
||||
[block.type]: '',
|
||||
},
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Create thinking_delta event
|
||||
*/
|
||||
createThinkingDeltaEvent(block: AccumulatorBlock, delta: string): AnthropicSSEEvent {
|
||||
return {
|
||||
event: 'content_block_delta',
|
||||
data: {
|
||||
type: 'content_block_delta',
|
||||
index: block.index,
|
||||
delta: {
|
||||
type: 'thinking_delta',
|
||||
thinking: delta,
|
||||
},
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Create text_delta event
|
||||
*/
|
||||
createTextDeltaEvent(block: AccumulatorBlock, delta: string): AnthropicSSEEvent {
|
||||
return {
|
||||
event: 'content_block_delta',
|
||||
data: {
|
||||
type: 'content_block_delta',
|
||||
index: block.index,
|
||||
delta: {
|
||||
type: 'text_delta',
|
||||
text: delta,
|
||||
},
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Create thinking signature delta event
|
||||
*/
|
||||
createSignatureDeltaEvent(block: AccumulatorBlock): AnthropicSSEEvent | null {
|
||||
// FIX: Guard against empty content (signature timing race)
|
||||
if (!block.content || block.content.length === 0) {
|
||||
if (this.verbose) {
|
||||
console.error(
|
||||
`[ResponseBuilder] WARNING: Skipping signature for empty thinking block ${block.index}`
|
||||
);
|
||||
console.error(
|
||||
`This indicates a race condition - signature requested before content accumulated`
|
||||
);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
const signature = this.generateThinkingSignature(block.content);
|
||||
|
||||
if (this.verbose) {
|
||||
console.error(
|
||||
`[ResponseBuilder] Generating signature for block ${block.index}: ${block.content.length} chars`
|
||||
);
|
||||
}
|
||||
|
||||
return {
|
||||
event: 'content_block_delta',
|
||||
data: {
|
||||
type: 'content_block_delta',
|
||||
index: block.index,
|
||||
delta: {
|
||||
type: 'thinking_signature_delta',
|
||||
signature: signature,
|
||||
},
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Create content_block_stop event
|
||||
*/
|
||||
createContentBlockStopEvent(block: AccumulatorBlock): AnthropicSSEEvent {
|
||||
return {
|
||||
event: 'content_block_stop',
|
||||
data: {
|
||||
type: 'content_block_stop',
|
||||
index: block.index,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Generate finalization events (message_delta + message_stop)
|
||||
*/
|
||||
createFinalizationEvents(accumulator: DeltaAccumulator, stopReason: string): AnthropicSSEEvent[] {
|
||||
return [
|
||||
{
|
||||
event: 'message_delta',
|
||||
data: {
|
||||
type: 'message_delta',
|
||||
delta: {
|
||||
stop_reason: stopReason,
|
||||
},
|
||||
usage: {
|
||||
input_tokens: accumulator.getInputTokens(),
|
||||
output_tokens: accumulator.getOutputTokens(),
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
event: 'message_stop',
|
||||
data: {
|
||||
type: 'message_stop',
|
||||
},
|
||||
},
|
||||
];
|
||||
}
|
||||
|
||||
/**
|
||||
* Generate thinking signature for Claude Code UI
|
||||
*/
|
||||
generateThinkingSignature(thinking: string): ThinkingSignature {
|
||||
// Generate signature hash
|
||||
const hash = crypto.createHash('sha256').update(thinking).digest('hex').substring(0, 16);
|
||||
|
||||
return {
|
||||
type: 'thinking_signature',
|
||||
hash: hash,
|
||||
length: thinking.length,
|
||||
timestamp: Date.now(),
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Map OpenAI stop reason to Anthropic stop reason
|
||||
*/
|
||||
mapStopReason(openaiReason: string): string {
|
||||
const mapping: Record<string, string> = {
|
||||
stop: 'end_turn',
|
||||
length: 'max_tokens',
|
||||
tool_calls: 'tool_use',
|
||||
content_filter: 'stop_sequence',
|
||||
};
|
||||
return mapping[openaiReason] || 'end_turn';
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,308 @@
|
||||
/**
|
||||
* StreamParser - Transform OpenAI streaming deltas to Anthropic SSE events
|
||||
*
|
||||
* Responsibilities:
|
||||
* - Process streaming deltas (reasoning_content, content, tool_calls)
|
||||
* - Coordinate with accumulator for state tracking
|
||||
* - Detect and handle planning loops
|
||||
* - Generate appropriate Anthropic SSE events
|
||||
*/
|
||||
|
||||
import type { DeltaAccumulator } from '../delta-accumulator';
|
||||
import type { SSEEvent, AnthropicSSEEvent, AccumulatorBlock } from './types';
|
||||
import { ResponseBuilder } from './response-builder';
|
||||
import { ToolCallHandler } from './tool-call-handler';
|
||||
|
||||
export interface StreamParserConfig {
|
||||
verbose?: boolean;
|
||||
debugMode?: boolean;
|
||||
debugLog?: boolean;
|
||||
writeDebugLog?: (type: string, data: unknown) => void;
|
||||
}
|
||||
|
||||
export class StreamParser {
|
||||
private verbose: boolean;
|
||||
private debugMode: boolean;
|
||||
private debugLog: boolean;
|
||||
private responseBuilder: ResponseBuilder;
|
||||
private toolCallHandler: ToolCallHandler;
|
||||
private writeDebugLog: (type: string, data: unknown) => void;
|
||||
|
||||
constructor(config: StreamParserConfig = {}) {
|
||||
this.verbose = config.verbose || false;
|
||||
this.debugMode = config.debugMode || false;
|
||||
this.debugLog = config.debugLog || false;
|
||||
this.responseBuilder = new ResponseBuilder(this.verbose);
|
||||
this.toolCallHandler = new ToolCallHandler();
|
||||
this.writeDebugLog = config.writeDebugLog || (() => {});
|
||||
}
|
||||
|
||||
/**
|
||||
* Transform OpenAI streaming delta to Anthropic events
|
||||
*/
|
||||
transformDelta(openaiEvent: SSEEvent, accumulator: DeltaAccumulator): AnthropicSSEEvent[] {
|
||||
const events: AnthropicSSEEvent[] = [];
|
||||
|
||||
// Debug logging for streaming deltas
|
||||
if (this.debugLog && openaiEvent.data) {
|
||||
this.writeDebugLog('delta-openai', openaiEvent.data);
|
||||
}
|
||||
|
||||
// Handle [DONE] marker
|
||||
if (openaiEvent.event === 'done') {
|
||||
if (!accumulator.isFinalized()) {
|
||||
return this.finalizeDelta(accumulator);
|
||||
}
|
||||
return []; // Already finalized
|
||||
}
|
||||
|
||||
// Usage update (appears in final chunk, may be before choice data)
|
||||
if (openaiEvent.data?.usage) {
|
||||
accumulator.updateUsage(openaiEvent.data.usage);
|
||||
|
||||
// If we have both usage AND finish_reason, finalize immediately
|
||||
if (accumulator.getFinishReason()) {
|
||||
events.push(...this.finalizeDelta(accumulator));
|
||||
return events;
|
||||
}
|
||||
}
|
||||
|
||||
const choice = openaiEvent.data?.choices?.[0];
|
||||
if (!choice) return events;
|
||||
|
||||
const delta = choice.delta;
|
||||
if (!delta) return events;
|
||||
|
||||
// Message start
|
||||
if (!accumulator.isMessageStarted()) {
|
||||
if (openaiEvent.data?.model) {
|
||||
accumulator.setModel(openaiEvent.data.model);
|
||||
}
|
||||
events.push(this.responseBuilder.createMessageStartEvent(accumulator));
|
||||
accumulator.setMessageStarted(true);
|
||||
}
|
||||
|
||||
// Role
|
||||
if (delta.role) {
|
||||
accumulator.setRole(delta.role);
|
||||
}
|
||||
|
||||
// Reasoning content delta
|
||||
if (delta.reasoning_content) {
|
||||
events.push(...this.handleReasoningDelta(delta.reasoning_content, accumulator));
|
||||
}
|
||||
|
||||
// Text content delta
|
||||
if (delta.content) {
|
||||
events.push(...this.handleContentDelta(delta.content, accumulator));
|
||||
}
|
||||
|
||||
// Check for planning loop
|
||||
if (accumulator.checkForLoop()) {
|
||||
this.log(
|
||||
'WARNING: Planning loop detected - 3 consecutive thinking blocks with no tool calls'
|
||||
);
|
||||
this.log('Forcing early finalization to prevent unbounded planning');
|
||||
events.push(...this.forceFinalization(accumulator));
|
||||
return events;
|
||||
}
|
||||
|
||||
// Tool calls deltas
|
||||
if (delta.tool_calls && delta.tool_calls.length > 0) {
|
||||
events.push(...this.handleToolCallDeltas(delta.tool_calls, accumulator));
|
||||
}
|
||||
|
||||
// Finish reason
|
||||
if (choice.finish_reason) {
|
||||
accumulator.setFinishReason(choice.finish_reason);
|
||||
|
||||
// If we have both finish_reason AND usage, finalize immediately
|
||||
if (accumulator.hasUsageReceived()) {
|
||||
events.push(...this.finalizeDelta(accumulator));
|
||||
}
|
||||
}
|
||||
|
||||
// Debug logging for generated events
|
||||
if (this.debugLog && events.length > 0) {
|
||||
this.writeDebugLog('delta-anthropic-events', {
|
||||
events,
|
||||
accumulator: accumulator.getSummary(),
|
||||
});
|
||||
}
|
||||
|
||||
return events;
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle reasoning content delta
|
||||
*/
|
||||
private handleReasoningDelta(
|
||||
reasoningContent: string,
|
||||
accumulator: DeltaAccumulator
|
||||
): AnthropicSSEEvent[] {
|
||||
const events: AnthropicSSEEvent[] = [];
|
||||
const currentBlock = accumulator.getCurrentBlock();
|
||||
|
||||
if (this.debugMode) {
|
||||
console.error(`[StreamParser] Reasoning delta: ${reasoningContent.length} chars`);
|
||||
console.error(
|
||||
`[StreamParser] Current block: ${currentBlock?.type || 'none'}, index: ${currentBlock?.index ?? 'N/A'}`
|
||||
);
|
||||
}
|
||||
|
||||
if (!currentBlock || currentBlock.type !== 'thinking') {
|
||||
// Start thinking block
|
||||
const block = accumulator.startBlock('thinking');
|
||||
events.push(this.responseBuilder.createContentBlockStartEvent(block));
|
||||
|
||||
if (this.debugMode) {
|
||||
console.error(`[StreamParser] Started new thinking block ${block.index}`);
|
||||
}
|
||||
}
|
||||
|
||||
accumulator.addDelta(reasoningContent);
|
||||
const currentThinkingBlock = accumulator.getCurrentBlock();
|
||||
if (currentThinkingBlock) {
|
||||
events.push(
|
||||
this.responseBuilder.createThinkingDeltaEvent(currentThinkingBlock, reasoningContent)
|
||||
);
|
||||
}
|
||||
|
||||
return events;
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle content delta
|
||||
*/
|
||||
private handleContentDelta(content: string, accumulator: DeltaAccumulator): AnthropicSSEEvent[] {
|
||||
const events: AnthropicSSEEvent[] = [];
|
||||
const currentBlock = accumulator.getCurrentBlock();
|
||||
|
||||
// Close thinking block if transitioning from thinking to text
|
||||
if (currentBlock && currentBlock.type === 'thinking' && !currentBlock.stopped) {
|
||||
events.push(...this.closeThinkingBlock(currentBlock, accumulator));
|
||||
}
|
||||
|
||||
if (!accumulator.getCurrentBlock() || accumulator.getCurrentBlock()?.type !== 'text') {
|
||||
// Start text block
|
||||
const block = accumulator.startBlock('text');
|
||||
events.push(this.responseBuilder.createContentBlockStartEvent(block));
|
||||
}
|
||||
|
||||
accumulator.addDelta(content);
|
||||
const currentTextBlock = accumulator.getCurrentBlock();
|
||||
if (currentTextBlock) {
|
||||
events.push(this.responseBuilder.createTextDeltaEvent(currentTextBlock, content));
|
||||
}
|
||||
|
||||
return events;
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle tool call deltas
|
||||
*/
|
||||
private handleToolCallDeltas(
|
||||
toolCallDeltas: import('./types').OpenAIToolCallDelta[],
|
||||
accumulator: DeltaAccumulator
|
||||
): AnthropicSSEEvent[] {
|
||||
const events: AnthropicSSEEvent[] = [];
|
||||
|
||||
if (!toolCallDeltas) return events;
|
||||
|
||||
// Close current content block ONCE before processing any tool calls
|
||||
const currentBlock = accumulator.getCurrentBlock();
|
||||
if (currentBlock && !currentBlock.stopped) {
|
||||
if (currentBlock.type === 'thinking') {
|
||||
events.push(...this.closeThinkingBlock(currentBlock, accumulator));
|
||||
} else {
|
||||
events.push(this.responseBuilder.createContentBlockStopEvent(currentBlock));
|
||||
accumulator.stopCurrentBlock();
|
||||
}
|
||||
}
|
||||
|
||||
// Process tool call deltas
|
||||
events.push(...this.toolCallHandler.processToolCallDeltas(toolCallDeltas, accumulator));
|
||||
|
||||
return events;
|
||||
}
|
||||
|
||||
/**
|
||||
* Close thinking block with signature
|
||||
*/
|
||||
private closeThinkingBlock(
|
||||
block: AccumulatorBlock,
|
||||
accumulator: DeltaAccumulator
|
||||
): AnthropicSSEEvent[] {
|
||||
const events: AnthropicSSEEvent[] = [];
|
||||
|
||||
const signatureEvent = this.responseBuilder.createSignatureDeltaEvent(block);
|
||||
if (signatureEvent) {
|
||||
events.push(signatureEvent);
|
||||
}
|
||||
events.push(this.responseBuilder.createContentBlockStopEvent(block));
|
||||
accumulator.stopCurrentBlock();
|
||||
|
||||
return events;
|
||||
}
|
||||
|
||||
/**
|
||||
* Force finalization due to loop detection
|
||||
*/
|
||||
private forceFinalization(accumulator: DeltaAccumulator): AnthropicSSEEvent[] {
|
||||
const events: AnthropicSSEEvent[] = [];
|
||||
|
||||
// Close current block if any
|
||||
const currentBlock = accumulator.getCurrentBlock();
|
||||
if (currentBlock && !currentBlock.stopped) {
|
||||
if (currentBlock.type === 'thinking') {
|
||||
events.push(...this.closeThinkingBlock(currentBlock, accumulator));
|
||||
} else {
|
||||
events.push(this.responseBuilder.createContentBlockStopEvent(currentBlock));
|
||||
accumulator.stopCurrentBlock();
|
||||
}
|
||||
}
|
||||
|
||||
// Force finalization
|
||||
events.push(...this.finalizeDelta(accumulator));
|
||||
return events;
|
||||
}
|
||||
|
||||
/**
|
||||
* Finalize streaming and generate closing events
|
||||
*/
|
||||
finalizeDelta(accumulator: DeltaAccumulator): AnthropicSSEEvent[] {
|
||||
if (accumulator.isFinalized()) {
|
||||
return []; // Already finalized
|
||||
}
|
||||
|
||||
const events: AnthropicSSEEvent[] = [];
|
||||
|
||||
// Close current content block if any
|
||||
const currentBlock = accumulator.getCurrentBlock();
|
||||
if (currentBlock && !currentBlock.stopped) {
|
||||
if (currentBlock.type === 'thinking') {
|
||||
events.push(...this.closeThinkingBlock(currentBlock, accumulator));
|
||||
} else {
|
||||
events.push(this.responseBuilder.createContentBlockStopEvent(currentBlock));
|
||||
accumulator.stopCurrentBlock();
|
||||
}
|
||||
}
|
||||
|
||||
// Message delta (stop reason + usage) and message stop
|
||||
const stopReason = this.responseBuilder.mapStopReason(accumulator.getFinishReason() || 'stop');
|
||||
events.push(...this.responseBuilder.createFinalizationEvents(accumulator, stopReason));
|
||||
|
||||
accumulator.setFinalized(true);
|
||||
return events;
|
||||
}
|
||||
|
||||
/**
|
||||
* Log message if verbose
|
||||
*/
|
||||
private log(message: string): void {
|
||||
if (this.verbose) {
|
||||
const timestamp = new Date().toTimeString().split(' ')[0]; // HH:MM:SS
|
||||
console.error(`[StreamParser] [${timestamp}] ${message}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,97 @@
|
||||
/**
|
||||
* ToolCallHandler - Handle tool call processing for streaming responses
|
||||
*
|
||||
* Responsibilities:
|
||||
* - Process tool call deltas from OpenAI
|
||||
* - Generate tool_use content blocks for Anthropic format
|
||||
* - Handle input_json_delta events
|
||||
*/
|
||||
|
||||
import type { DeltaAccumulator } from '../delta-accumulator';
|
||||
import type { OpenAIToolCallDelta, OpenAIToolCall, ContentBlock, AnthropicSSEEvent } from './types';
|
||||
|
||||
export class ToolCallHandler {
|
||||
/**
|
||||
* Process tool calls from non-streaming response
|
||||
*/
|
||||
processToolCalls(toolCalls: OpenAIToolCall[]): ContentBlock[] {
|
||||
const content: ContentBlock[] = [];
|
||||
|
||||
for (const toolCall of toolCalls) {
|
||||
let parsedInput: Record<string, unknown>;
|
||||
try {
|
||||
parsedInput = JSON.parse(toolCall.function.arguments || '{}');
|
||||
} catch (parseError) {
|
||||
const err = parseError as Error;
|
||||
console.error(`[ToolCallHandler] Invalid JSON in tool arguments: ${err.message}`);
|
||||
parsedInput = { _error: 'Invalid JSON', _raw: toolCall.function.arguments };
|
||||
}
|
||||
|
||||
content.push({
|
||||
type: 'tool_use',
|
||||
id: toolCall.id,
|
||||
name: toolCall.function.name,
|
||||
input: parsedInput,
|
||||
});
|
||||
}
|
||||
|
||||
return content;
|
||||
}
|
||||
|
||||
/**
|
||||
* Process tool call deltas during streaming
|
||||
* Returns events for tool use blocks and input_json deltas
|
||||
*/
|
||||
processToolCallDeltas(
|
||||
toolCallDeltas: OpenAIToolCallDelta[],
|
||||
accumulator: DeltaAccumulator
|
||||
): AnthropicSSEEvent[] {
|
||||
const events: AnthropicSSEEvent[] = [];
|
||||
|
||||
for (const toolCallDelta of toolCallDeltas) {
|
||||
// Track tool call state
|
||||
const isNewToolCall = !accumulator.hasToolCall(toolCallDelta.index);
|
||||
accumulator.addToolCallDelta(toolCallDelta);
|
||||
|
||||
// Emit tool use events (start + input_json deltas)
|
||||
if (isNewToolCall) {
|
||||
// Start new tool_use block in accumulator
|
||||
const block = accumulator.startBlock('tool_use');
|
||||
const toolCall = accumulator.getToolCall(toolCallDelta.index);
|
||||
|
||||
events.push({
|
||||
event: 'content_block_start',
|
||||
data: {
|
||||
type: 'content_block_start',
|
||||
index: block.index,
|
||||
content_block: {
|
||||
type: 'tool_use',
|
||||
id: toolCall?.id || `tool_${toolCallDelta.index}`,
|
||||
name: toolCall?.function?.name || '',
|
||||
},
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
// Emit input_json delta if arguments present
|
||||
if (toolCallDelta.function?.arguments) {
|
||||
const currentToolBlock = accumulator.getCurrentBlock();
|
||||
if (currentToolBlock && currentToolBlock.type === 'tool_use') {
|
||||
events.push({
|
||||
event: 'content_block_delta',
|
||||
data: {
|
||||
type: 'content_block_delta',
|
||||
index: currentToolBlock.index,
|
||||
delta: {
|
||||
type: 'input_json_delta',
|
||||
partial_json: toolCallDelta.function.arguments,
|
||||
},
|
||||
},
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return events;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,185 @@
|
||||
/**
|
||||
* Pipeline Types - Shared types for GLMT transformation pipeline
|
||||
*/
|
||||
|
||||
// Content block types
|
||||
export interface ContentBlock {
|
||||
type: string;
|
||||
text?: string;
|
||||
thinking?: string;
|
||||
signature?: ThinkingSignature;
|
||||
id?: string;
|
||||
name?: string;
|
||||
input?: Record<string, unknown>;
|
||||
tool_use_id?: string;
|
||||
content?: string | unknown;
|
||||
}
|
||||
|
||||
export interface Message {
|
||||
role: string;
|
||||
content: string | ContentBlock[];
|
||||
}
|
||||
|
||||
// Tool types
|
||||
export interface AnthropicTool {
|
||||
name: string;
|
||||
description: string;
|
||||
input_schema: Record<string, unknown>;
|
||||
}
|
||||
|
||||
export interface OpenAITool {
|
||||
type: 'function';
|
||||
function: {
|
||||
name: string;
|
||||
description: string;
|
||||
parameters: Record<string, unknown>;
|
||||
};
|
||||
}
|
||||
|
||||
export interface OpenAIToolCall {
|
||||
id: string;
|
||||
type: string;
|
||||
function: {
|
||||
name: string;
|
||||
arguments: string;
|
||||
};
|
||||
}
|
||||
|
||||
export interface OpenAIToolCallDelta {
|
||||
index: number;
|
||||
id?: string;
|
||||
type?: string;
|
||||
function?: {
|
||||
name?: string;
|
||||
arguments?: string;
|
||||
};
|
||||
}
|
||||
|
||||
// Request/Response types
|
||||
export interface AnthropicRequest {
|
||||
model: string;
|
||||
messages: Message[];
|
||||
max_tokens?: number;
|
||||
temperature?: number;
|
||||
top_p?: number;
|
||||
tools?: AnthropicTool[];
|
||||
stream?: boolean;
|
||||
thinking?: {
|
||||
type: 'enabled' | 'disabled';
|
||||
budget_tokens?: number;
|
||||
};
|
||||
}
|
||||
|
||||
export interface OpenAIRequest {
|
||||
model: string;
|
||||
messages: Message[];
|
||||
max_tokens?: number;
|
||||
temperature?: number;
|
||||
top_p?: number;
|
||||
tools?: OpenAITool[];
|
||||
tool_choice?: string;
|
||||
stream?: boolean;
|
||||
do_sample?: boolean;
|
||||
reasoning?: boolean;
|
||||
reasoning_effort?: string;
|
||||
}
|
||||
|
||||
export interface ThinkingConfig {
|
||||
thinking: boolean;
|
||||
effort: string;
|
||||
}
|
||||
|
||||
export interface TransformResult {
|
||||
openaiRequest: OpenAIRequest | AnthropicRequest;
|
||||
thinkingConfig: ThinkingConfig;
|
||||
error?: string;
|
||||
}
|
||||
|
||||
export interface ThinkingSignature {
|
||||
type: string;
|
||||
hash: string;
|
||||
length: number;
|
||||
timestamp: number;
|
||||
}
|
||||
|
||||
// OpenAI response types
|
||||
export interface OpenAIChoice {
|
||||
message: {
|
||||
role?: string;
|
||||
content?: string | null;
|
||||
reasoning_content?: string;
|
||||
tool_calls?: OpenAIToolCall[];
|
||||
};
|
||||
delta?: {
|
||||
role?: string;
|
||||
content?: string;
|
||||
reasoning_content?: string;
|
||||
tool_calls?: OpenAIToolCallDelta[];
|
||||
};
|
||||
finish_reason?: string;
|
||||
}
|
||||
|
||||
export interface OpenAIResponse {
|
||||
id?: string;
|
||||
model?: string;
|
||||
choices?: OpenAIChoice[];
|
||||
usage?: {
|
||||
prompt_tokens?: number;
|
||||
completion_tokens?: number;
|
||||
};
|
||||
}
|
||||
|
||||
export interface AnthropicResponse {
|
||||
id: string;
|
||||
type: string;
|
||||
role: string;
|
||||
content: ContentBlock[];
|
||||
model: string;
|
||||
stop_reason: string;
|
||||
usage: {
|
||||
input_tokens: number;
|
||||
output_tokens: number;
|
||||
};
|
||||
}
|
||||
|
||||
// SSE event types
|
||||
export interface SSEEvent {
|
||||
event?: string;
|
||||
data?: OpenAIResponse & {
|
||||
usage?: {
|
||||
prompt_tokens?: number;
|
||||
completion_tokens?: number;
|
||||
};
|
||||
};
|
||||
}
|
||||
|
||||
export interface AnthropicSSEEvent {
|
||||
event: string;
|
||||
data: Record<string, unknown>;
|
||||
}
|
||||
|
||||
// Accumulator block type
|
||||
export interface AccumulatorBlock {
|
||||
index: number;
|
||||
type: string;
|
||||
content: string;
|
||||
stopped: boolean;
|
||||
}
|
||||
|
||||
// Validation result
|
||||
export interface ValidationResult {
|
||||
checks: Record<string, boolean>;
|
||||
passed: number;
|
||||
total: number;
|
||||
valid: boolean;
|
||||
}
|
||||
|
||||
// Config types
|
||||
export interface GlmtTransformerConfig {
|
||||
defaultThinking?: boolean;
|
||||
verbose?: boolean;
|
||||
debugLog?: boolean;
|
||||
debugMode?: boolean;
|
||||
debugLogDir?: string;
|
||||
explicitReasoning?: boolean;
|
||||
}
|
||||
Reference in New Issue
Block a user