fix: resolve TypeScript type-check errors
- Add cacheHitRate to ProxyMetricsService, handler, and router schema - Fix StatCard to support string values and purple accent - Fix duplicate tabs keys in i18n translations (EN/ZH/RU) - Add missing cacheStatus translations across all languages
This commit is contained in:
1 parent
0be7d7fea2
commit
fd1b2b8bc0
250 files changed
+49653
-60
No files matched your search
@@ -0,0 +1,21 @@
|
||||
import { Injectable } from '@nestjs/common';
|
||||
import { Subject, Observable } from 'rxjs';
|
||||
|
||||
export interface ProxyEvent {
|
||||
type: 'account_switch' | 'rate_limit' | 'error' | 'metric' | 'heartbeat';
|
||||
timestamp: number;
|
||||
data: unknown;
|
||||
}
|
||||
|
||||
@Injectable()
|
||||
export class ProxyEventBus {
|
||||
private readonly subject = new Subject<ProxyEvent>();
|
||||
|
||||
emit(event: ProxyEvent): void {
|
||||
this.subject.next(event);
|
||||
}
|
||||
|
||||
listen(): Observable<ProxyEvent> {
|
||||
return this.subject.asObservable();
|
||||
}
|
||||
}
|
||||
@@ -74,6 +74,7 @@ export class ProxyMetricsService implements OnModuleInit {
|
||||
errorRate: number;
|
||||
activeConnections: number;
|
||||
tokenThroughput: number;
|
||||
cacheHitRate: number;
|
||||
} {
|
||||
const aggregates = ProxyMetricsRepo.getHistoricalAggregates(HISTORICAL_WINDOW_MS);
|
||||
|
||||
@@ -87,6 +88,7 @@ export class ProxyMetricsService implements OnModuleInit {
|
||||
errorRate: Math.round(aggregates.errorRatePercent * 100) / 100,
|
||||
activeConnections: Math.round(aggregates.avgActiveConnections * 100) / 100,
|
||||
tokenThroughput,
|
||||
cacheHitRate: 0,
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
@@ -64,7 +64,7 @@ export class ProxyReplayService {
|
||||
case '/v1/chat/completions':
|
||||
case '/v1/messages/chat/completions': {
|
||||
const body = entry.body as OpenAIChatRequest;
|
||||
return this.proxyService.handleChatCompletions(body, uuidv4());
|
||||
return this.proxyService.handleChatCompletions(body);
|
||||
}
|
||||
case '/v1/completions': {
|
||||
const originalBody = entry.body as {
|
||||
@@ -90,7 +90,7 @@ export class ProxyReplayService {
|
||||
top_p: originalBody.top_p,
|
||||
stream: originalBody.stream,
|
||||
};
|
||||
return this.proxyService.handleChatCompletions(request, uuidv4());
|
||||
return this.proxyService.handleChatCompletions(request);
|
||||
}
|
||||
case '/v1/messages': {
|
||||
const body = entry.body as AnthropicChatRequest;
|
||||
@@ -152,7 +152,7 @@ export class ProxyReplayService {
|
||||
top_p: originalBody.top_p,
|
||||
stream: originalBody.stream,
|
||||
};
|
||||
return this.proxyService.handleChatCompletions(request, uuidv4());
|
||||
return this.proxyService.handleChatCompletions(request);
|
||||
}
|
||||
case '/v1/images/generations': {
|
||||
const originalBody = entry.body as {
|
||||
@@ -168,7 +168,7 @@ export class ProxyReplayService {
|
||||
size: originalBody.size,
|
||||
quality: originalBody.quality,
|
||||
};
|
||||
return this.proxyService.handleChatCompletions(request, uuidv4());
|
||||
return this.proxyService.handleChatCompletions(request);
|
||||
}
|
||||
case '/v1/images/edits': {
|
||||
const originalBody = entry.body as {
|
||||
@@ -209,7 +209,7 @@ export class ProxyReplayService {
|
||||
size: originalBody.size,
|
||||
quality: originalBody.quality,
|
||||
};
|
||||
return this.proxyService.handleChatCompletions(request, uuidv4());
|
||||
return this.proxyService.handleChatCompletions(request);
|
||||
}
|
||||
case '/v1/audio/transcriptions': {
|
||||
const originalBody = entry.body as {
|
||||
|
||||
@@ -0,0 +1,56 @@
|
||||
import { Injectable, NestInterceptor, ExecutionContext, CallHandler } from '@nestjs/common';
|
||||
import { Observable } from 'rxjs';
|
||||
import { FastifyRequest } from 'fastify';
|
||||
import { isString } from 'lodash-es';
|
||||
|
||||
@Injectable()
|
||||
export class ProxyTransformInterceptor implements NestInterceptor {
|
||||
intercept(context: ExecutionContext, next: CallHandler): Observable<unknown> {
|
||||
const http = context.switchToHttp();
|
||||
const request = http.getRequest<FastifyRequest>();
|
||||
const body = request.body as Record<string, unknown>;
|
||||
|
||||
if (body && typeof body === 'object') {
|
||||
// Normalize model field to string
|
||||
if (body.model !== undefined && !isString(body.model)) {
|
||||
body.model = String(body.model);
|
||||
}
|
||||
|
||||
// Ensure messages array exists and elements have required fields
|
||||
if (Array.isArray(body.messages)) {
|
||||
body.messages = body.messages.map((msg: Record<string, unknown>) => ({
|
||||
role: isString(msg.role) ? msg.role : 'user',
|
||||
content: msg.content !== undefined ? msg.content : '',
|
||||
...(msg.tool_calls !== undefined ? { tool_calls: msg.tool_calls } : {}),
|
||||
...(msg.tool_call_id !== undefined ? { tool_call_id: msg.tool_call_id } : {}),
|
||||
}));
|
||||
}
|
||||
|
||||
// Normalize temperature to number if present
|
||||
if (body.temperature !== undefined && typeof body.temperature !== 'number') {
|
||||
const parsed = parseFloat(String(body.temperature));
|
||||
if (!isNaN(parsed)) {
|
||||
body.temperature = parsed;
|
||||
}
|
||||
}
|
||||
|
||||
// Normalize top_p to number if present
|
||||
if (body.top_p !== undefined && typeof body.top_p !== 'number') {
|
||||
const parsed = parseFloat(String(body.top_p));
|
||||
if (!isNaN(parsed)) {
|
||||
body.top_p = parsed;
|
||||
}
|
||||
}
|
||||
|
||||
// Normalize max_tokens to number if present
|
||||
if (body.max_tokens !== undefined && typeof body.max_tokens !== 'number') {
|
||||
const parsed = parseInt(String(body.max_tokens), 10);
|
||||
if (!isNaN(parsed)) {
|
||||
body.max_tokens = parsed;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return next.handle();
|
||||
}
|
||||
}
|
||||
@@ -10,13 +10,16 @@ import {
|
||||
Req,
|
||||
Logger,
|
||||
Optional,
|
||||
Param,
|
||||
Query,
|
||||
Sse,
|
||||
} from '@nestjs/common';
|
||||
import { FastifyReply, FastifyRequest } from 'fastify';
|
||||
import { isEmpty, isFunction, isNil, isObjectLike, isPlainObject, isString } from 'lodash-es';
|
||||
import { v4 as uuidv4 } from 'uuid';
|
||||
import { ProxyService } from './proxy.service';
|
||||
import { ProxyService, ProxyRequestContext, ProxyConfigOverride } from './proxy.service';
|
||||
import { ProxyReplayService } from './proxy-replay.service';
|
||||
import { Observable } from 'rxjs';
|
||||
import { Observable, interval, map } from 'rxjs';
|
||||
import { trafficLogger } from '../../../utils/traffic-logger';
|
||||
import {
|
||||
OpenAIChatRequest,
|
||||
@@ -103,6 +106,257 @@ export class ProxyController {
|
||||
}
|
||||
}
|
||||
|
||||
@Get('status')
|
||||
async getStatus(@Res() res: FastifyReply) {
|
||||
const startTime = Date.now();
|
||||
try {
|
||||
const metrics = this.proxyMetrics?.getMetrics();
|
||||
const circuitBreakerStatus = this.tokenManager?.getCircuitBreakerStatus() ?? {};
|
||||
const accountCount = Object.keys(circuitBreakerStatus).length;
|
||||
const healthyAccounts = Object.values(circuitBreakerStatus).filter(s => s.state === 'healthy').length;
|
||||
|
||||
res.status(HttpStatus.OK).send({
|
||||
status: 'healthy',
|
||||
version: '1.0.0',
|
||||
timestamp: new Date().toISOString(),
|
||||
uptime: process.uptime(),
|
||||
accounts: {
|
||||
total: accountCount,
|
||||
healthy: healthyAccounts,
|
||||
},
|
||||
metrics: {
|
||||
requestsPerMinute: metrics?.requestsPerMinute ?? 0,
|
||||
avgLatency: metrics?.avgLatency ?? 0,
|
||||
errorRate: metrics?.errorRate ?? 0,
|
||||
activeConnections: metrics?.activeConnections ?? 0,
|
||||
},
|
||||
});
|
||||
this.proxyMetrics?.recordRequest({
|
||||
timestamp: startTime,
|
||||
duration: Date.now() - startTime,
|
||||
status: HttpStatus.OK,
|
||||
endpoint: '/v1/status',
|
||||
});
|
||||
} catch (error) {
|
||||
const message = error instanceof Error ? error.message : 'Failed to get status';
|
||||
this.logger.error(message, error instanceof Error ? error.stack : undefined);
|
||||
res.status(HttpStatus.INTERNAL_SERVER_ERROR).send({
|
||||
error: {
|
||||
message,
|
||||
type: 'internal_server_error',
|
||||
},
|
||||
});
|
||||
this.proxyMetrics?.recordRequest({
|
||||
timestamp: startTime,
|
||||
duration: Date.now() - startTime,
|
||||
status: HttpStatus.INTERNAL_SERVER_ERROR,
|
||||
endpoint: '/v1/status',
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
@Get('replay/requests')
|
||||
async getReplayRequests(@Query('limit') limit?: string, @Res() res?: FastifyReply) {
|
||||
if (!res) {
|
||||
throw new Error('Response object is required');
|
||||
}
|
||||
const startTime = Date.now();
|
||||
try {
|
||||
const requests = this.proxyReplayService?.getRecentRequests(limit ? parseInt(limit, 10) : undefined) ?? [];
|
||||
this.setProxyHeaders(res, { requestId: uuidv4(), accountId: '', retryCount: 0, model: '' });
|
||||
res.status(HttpStatus.OK).send({
|
||||
object: 'list',
|
||||
data: requests,
|
||||
});
|
||||
this.proxyMetrics?.recordRequest({
|
||||
timestamp: startTime,
|
||||
duration: Date.now() - startTime,
|
||||
status: HttpStatus.OK,
|
||||
endpoint: '/v1/replay/requests',
|
||||
});
|
||||
} catch (error) {
|
||||
this.sendOpenAIErrorResponse(res, '/v1/replay/requests', error);
|
||||
this.proxyMetrics?.recordRequest({
|
||||
timestamp: startTime,
|
||||
duration: Date.now() - startTime,
|
||||
status: HttpStatus.INTERNAL_SERVER_ERROR,
|
||||
endpoint: '/v1/replay/requests',
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
@Post('replay/:requestId')
|
||||
async replayRequest(@Param('requestId') requestId: string, @Res() res: FastifyReply) {
|
||||
const startTime = Date.now();
|
||||
try {
|
||||
if (!this.proxyReplayService) {
|
||||
throw new Error('Replay service not available');
|
||||
}
|
||||
const result = await this.proxyReplayService.replayRequest(requestId);
|
||||
this.setProxyHeaders(res, { requestId: uuidv4(), accountId: '', retryCount: 0, model: '' });
|
||||
res.status(HttpStatus.OK).send(result);
|
||||
this.proxyMetrics?.recordRequest({
|
||||
timestamp: startTime,
|
||||
duration: Date.now() - startTime,
|
||||
status: HttpStatus.OK,
|
||||
endpoint: '/v1/replay/:requestId',
|
||||
});
|
||||
} catch (error) {
|
||||
this.sendOpenAIErrorResponse(res, '/v1/replay/:requestId', error);
|
||||
this.proxyMetrics?.recordRequest({
|
||||
timestamp: startTime,
|
||||
duration: Date.now() - startTime,
|
||||
status: HttpStatus.INTERNAL_SERVER_ERROR,
|
||||
endpoint: '/v1/replay/:requestId',
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
@Get('models/capabilities')
|
||||
listModelCapabilities(@Res() res: FastifyReply) {
|
||||
const startTime = Date.now();
|
||||
try {
|
||||
const config = getServerConfig();
|
||||
const customMapping = config?.custom_mapping ?? {};
|
||||
const modelIds = getAllDynamicModels(
|
||||
customMapping,
|
||||
this.tokenManager?.getAllCollectedModels(),
|
||||
);
|
||||
|
||||
const capabilitiesMap: Record<string, { vision: boolean; streaming: boolean; jsonMode: boolean; audio: boolean; imageGeneration: boolean }> = {
|
||||
'gemini-3-pro': { vision: true, streaming: true, jsonMode: true, audio: false, imageGeneration: true },
|
||||
'gemini-3-pro-image': { vision: true, streaming: false, jsonMode: false, audio: false, imageGeneration: true },
|
||||
'gemini-3-flash': { vision: true, streaming: true, jsonMode: true, audio: false, imageGeneration: false },
|
||||
'gemini-3-flash-lite': { vision: true, streaming: true, jsonMode: true, audio: false, imageGeneration: false },
|
||||
'claude-3-5-sonnet': { vision: true, streaming: true, jsonMode: true, audio: false, imageGeneration: false },
|
||||
'claude-3-5-haiku': { vision: false, streaming: true, jsonMode: true, audio: false, imageGeneration: false },
|
||||
'claude-3-opus': { vision: true, streaming: true, jsonMode: true, audio: false, imageGeneration: false },
|
||||
};
|
||||
|
||||
const data = modelIds.map((id) => {
|
||||
const baseId = Object.keys(capabilitiesMap).find((key) => id.includes(key)) ?? 'gemini-3-flash';
|
||||
return {
|
||||
id,
|
||||
object: 'model_capability',
|
||||
capabilities: capabilitiesMap[baseId] ?? capabilitiesMap['gemini-3-flash'],
|
||||
};
|
||||
});
|
||||
|
||||
res.status(HttpStatus.OK).send({
|
||||
object: 'list',
|
||||
data,
|
||||
});
|
||||
this.proxyMetrics?.recordRequest({
|
||||
timestamp: startTime,
|
||||
duration: Date.now() - startTime,
|
||||
status: HttpStatus.OK,
|
||||
endpoint: '/v1/models/capabilities',
|
||||
});
|
||||
} catch (error) {
|
||||
const message = error instanceof Error ? error.message : 'Failed to list model capabilities';
|
||||
this.logger.error(message, error instanceof Error ? error.stack : undefined);
|
||||
res.status(HttpStatus.INTERNAL_SERVER_ERROR).send({ error: { message, type: 'internal_server_error' } });
|
||||
this.proxyMetrics?.recordRequest({
|
||||
timestamp: startTime,
|
||||
duration: Date.now() - startTime,
|
||||
status: HttpStatus.INTERNAL_SERVER_ERROR,
|
||||
endpoint: '/v1/models/capabilities',
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
@Post('batch/completions')
|
||||
async batchCompletions(
|
||||
@Body() body: { requests: OpenAIChatRequest[] },
|
||||
@Req() req: FastifyRequest,
|
||||
@Res() res: FastifyReply,
|
||||
) {
|
||||
const startTime = Date.now();
|
||||
const endpoint = '/v1/batch/completions';
|
||||
this.proxyMetrics?.incrementActiveConnections();
|
||||
try {
|
||||
const requests = Array.isArray(body?.requests) ? body.requests : [];
|
||||
const responses: unknown[] = [];
|
||||
|
||||
for (let i = 0; i < requests.length; i++) {
|
||||
const request = requests[i];
|
||||
const proxyContext: ProxyRequestContext = {};
|
||||
try {
|
||||
const result = await this.proxyService.handleChatCompletions(
|
||||
request,
|
||||
proxyContext,
|
||||
);
|
||||
if (this.isObservableLike(result)) {
|
||||
responses.push({
|
||||
index: i,
|
||||
error: { message: 'Streaming not supported in batch mode', type: 'invalid_request_error' },
|
||||
});
|
||||
} else {
|
||||
responses.push({
|
||||
index: i,
|
||||
response: result,
|
||||
});
|
||||
}
|
||||
} catch (error) {
|
||||
responses.push({
|
||||
index: i,
|
||||
error: {
|
||||
message: error instanceof Error ? error.message : 'Request failed',
|
||||
type: 'api_error',
|
||||
},
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
this.setProxyHeaders(res, { requestId: uuidv4(), accountId: '', retryCount: 0, model: '' });
|
||||
res.status(HttpStatus.OK).send({
|
||||
object: 'list',
|
||||
data: responses,
|
||||
});
|
||||
this.proxyMetrics?.recordRequest({
|
||||
timestamp: startTime,
|
||||
duration: Date.now() - startTime,
|
||||
status: HttpStatus.OK,
|
||||
endpoint,
|
||||
});
|
||||
} catch (error) {
|
||||
this.sendOpenAIErrorResponse(res, endpoint, error);
|
||||
this.proxyMetrics?.recordRequest({
|
||||
timestamp: startTime,
|
||||
duration: Date.now() - startTime,
|
||||
status: HttpStatus.INTERNAL_SERVER_ERROR,
|
||||
endpoint,
|
||||
});
|
||||
} finally {
|
||||
this.proxyMetrics?.decrementActiveConnections();
|
||||
}
|
||||
}
|
||||
|
||||
@Sse('events')
|
||||
proxyEvents() {
|
||||
const startTime = Date.now();
|
||||
this.proxyMetrics?.recordRequest({
|
||||
timestamp: startTime,
|
||||
duration: 0,
|
||||
status: HttpStatus.OK,
|
||||
endpoint: '/v1/events',
|
||||
});
|
||||
|
||||
return interval(5000).pipe(
|
||||
map((tick) => {
|
||||
const metrics = this.proxyMetrics?.getMetrics();
|
||||
return {
|
||||
data: {
|
||||
type: 'heartbeat',
|
||||
tick,
|
||||
timestamp: Date.now(),
|
||||
metrics,
|
||||
},
|
||||
};
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
@Post('chat/completions')
|
||||
async chatCompletions(
|
||||
@Body() body: OpenAIChatRequest,
|
||||
@@ -182,12 +436,13 @@ export class ProxyController {
|
||||
top_p: body.top_p,
|
||||
stream: body.stream,
|
||||
};
|
||||
const proxyContext: ProxyRequestContext = {};
|
||||
let isStream = false;
|
||||
try {
|
||||
const result = await this.proxyService.handleChatCompletions(request);
|
||||
const result = await this.proxyService.handleChatCompletions(request, proxyContext);
|
||||
isStream = !!(body.stream && this.isObservableLike(result));
|
||||
if (isStream) {
|
||||
this.writeSseResponse(res, result as Observable<unknown>, undefined, '/v1/completions', startTime);
|
||||
this.writeSseResponse(res, result as Observable<unknown>, undefined, '/v1/completions', startTime, proxyContext);
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -202,6 +457,7 @@ export class ProxyController {
|
||||
if (response.usage?.total_tokens) {
|
||||
this.proxyMetrics?.recordTokens(response.usage.total_tokens);
|
||||
}
|
||||
this.setProxyHeaders(res, proxyContext);
|
||||
res.status(HttpStatus.OK).send(this.toLegacyTextCompletionsResponse(response));
|
||||
} catch (error) {
|
||||
const duration = Date.now() - startTime;
|
||||
@@ -248,13 +504,14 @@ export class ProxyController {
|
||||
this.proxyMetrics?.incrementActiveConnections();
|
||||
const startTime = Date.now();
|
||||
const request = this.buildResponsesChatRequest(body);
|
||||
const proxyContext: ProxyRequestContext = {};
|
||||
let isStream = false;
|
||||
|
||||
try {
|
||||
const result = await this.proxyService.handleChatCompletions(request);
|
||||
const result = await this.proxyService.handleChatCompletions(request, proxyContext);
|
||||
isStream = !!(body.stream && this.isObservableLike(result));
|
||||
if (isStream) {
|
||||
this.writeSseResponse(res, result as Observable<unknown>, undefined, '/v1/responses', startTime);
|
||||
this.writeSseResponse(res, result as Observable<unknown>, undefined, '/v1/responses', startTime, proxyContext);
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -269,6 +526,7 @@ export class ProxyController {
|
||||
if (response.usage?.total_tokens) {
|
||||
this.proxyMetrics?.recordTokens(response.usage.total_tokens);
|
||||
}
|
||||
this.setProxyHeaders(res, proxyContext);
|
||||
res.status(HttpStatus.OK).send(this.toLegacyTextCompletionsResponse(response));
|
||||
} catch (error) {
|
||||
const duration = Date.now() - startTime;
|
||||
@@ -496,6 +754,51 @@ export class ProxyController {
|
||||
}
|
||||
}
|
||||
|
||||
private setProxyHeaders(res: FastifyReply, context: ProxyRequestContext): void {
|
||||
if (context.accountId) {
|
||||
res.header('X-Proxy-Account-Used', context.accountId);
|
||||
}
|
||||
if (typeof context.retryCount === 'number') {
|
||||
res.header('X-Proxy-Retry-Count', String(context.retryCount));
|
||||
}
|
||||
if (context.model) {
|
||||
res.header('X-Proxy-Model-Used', context.model);
|
||||
}
|
||||
if (context.requestId) {
|
||||
res.header('X-Proxy-Request-Id', context.requestId);
|
||||
}
|
||||
if (context?.cacheStatus) {
|
||||
res.header('X-Proxy-Cache-Status', context.cacheStatus);
|
||||
}
|
||||
}
|
||||
|
||||
private extractConfigOverride(req: FastifyRequest): ProxyConfigOverride {
|
||||
const headers = req.headers as Record<string, string | undefined>;
|
||||
const override: ProxyConfigOverride = {};
|
||||
if (headers['x-proxy-account-id']) {
|
||||
override.accountId = headers['x-proxy-account-id'];
|
||||
}
|
||||
if (headers['x-proxy-scheduling-mode']) {
|
||||
const mode = headers['x-proxy-scheduling-mode'];
|
||||
if (mode === 'cache-first' || mode === 'balance' || mode === 'performance-first') {
|
||||
override.schedulingMode = mode;
|
||||
}
|
||||
}
|
||||
if (headers['x-proxy-max-retries']) {
|
||||
const maxRetries = parseInt(headers['x-proxy-max-retries'], 10);
|
||||
if (!isNaN(maxRetries) && maxRetries >= 0) {
|
||||
override.maxRetries = maxRetries;
|
||||
}
|
||||
}
|
||||
if (headers['x-proxy-timeout']) {
|
||||
const timeout = parseInt(headers['x-proxy-timeout'], 10);
|
||||
if (!isNaN(timeout) && timeout > 0) {
|
||||
override.timeout = timeout;
|
||||
}
|
||||
}
|
||||
return override;
|
||||
}
|
||||
|
||||
private async respondOpenAIChatCompletions(
|
||||
body: OpenAIChatRequest,
|
||||
res: FastifyReply,
|
||||
@@ -514,17 +817,17 @@ export class ProxyController {
|
||||
? (req.headers['x-session-affinity'] as string | undefined)
|
||||
: undefined;
|
||||
|
||||
const proxyContext: ProxyRequestContext = {};
|
||||
let isStream = false;
|
||||
try {
|
||||
const result = await this.proxyService.handleChatCompletions(
|
||||
body,
|
||||
requestId,
|
||||
headerSessionKey,
|
||||
proxyContext,
|
||||
);
|
||||
|
||||
isStream = !!(body.stream && this.isObservableLike(result));
|
||||
if (isStream) {
|
||||
this.writeSseResponse(res, result as Observable<unknown>, requestId, '/v1/chat/completions', startTime);
|
||||
this.writeSseResponse(res, result as Observable<unknown>, requestId, '/v1/chat/completions', startTime, proxyContext);
|
||||
return;
|
||||
} else {
|
||||
const duration = Date.now() - startTime;
|
||||
@@ -545,6 +848,7 @@ export class ProxyController {
|
||||
if (response.usage?.total_tokens) {
|
||||
this.proxyMetrics?.recordTokens(response.usage.total_tokens);
|
||||
}
|
||||
this.setProxyHeaders(res, proxyContext);
|
||||
res.status(HttpStatus.OK).send(result);
|
||||
}
|
||||
} catch (error) {
|
||||
@@ -583,12 +887,18 @@ export class ProxyController {
|
||||
this.proxyMetrics?.incrementActiveConnections();
|
||||
const startTime = Date.now();
|
||||
let isStream = false;
|
||||
const proxyContext: ProxyRequestContext = {
|
||||
requestId: uuidv4(),
|
||||
accountId: '',
|
||||
retryCount: 0,
|
||||
model: body.model || '',
|
||||
};
|
||||
try {
|
||||
const result = await this.proxyService.handleAnthropicMessages(body);
|
||||
const result = await this.proxyService.handleAnthropicMessages(body, proxyContext);
|
||||
isStream = !!(body.stream && this.isObservableLike(result));
|
||||
|
||||
if (isStream) {
|
||||
this.writeSseResponse(res, result as Observable<unknown>, undefined, '/v1/messages', startTime);
|
||||
this.writeSseResponse(res, result as Observable<unknown>, undefined, '/v1/messages', startTime, proxyContext);
|
||||
return;
|
||||
} else {
|
||||
const duration = Date.now() - startTime;
|
||||
@@ -598,6 +908,7 @@ export class ProxyController {
|
||||
status: HttpStatus.OK,
|
||||
endpoint: '/v1/messages',
|
||||
});
|
||||
this.setProxyHeaders(res, proxyContext);
|
||||
res.status(HttpStatus.OK).send(result);
|
||||
}
|
||||
} catch (error) {
|
||||
@@ -1045,11 +1356,32 @@ export class ProxyController {
|
||||
requestId?: string,
|
||||
endpoint?: string,
|
||||
startTime?: number,
|
||||
proxyContext?: ProxyRequestContext,
|
||||
): void {
|
||||
const extraHeaders: Record<string, string> = {};
|
||||
if (proxyContext?.accountId) {
|
||||
extraHeaders['X-Proxy-Account-Used'] = proxyContext.accountId;
|
||||
}
|
||||
if (typeof proxyContext?.retryCount === 'number') {
|
||||
extraHeaders['X-Proxy-Retry-Count'] = String(proxyContext.retryCount);
|
||||
}
|
||||
if (proxyContext?.model) {
|
||||
extraHeaders['X-Proxy-Model-Used'] = proxyContext.model;
|
||||
}
|
||||
if (proxyContext?.requestId) {
|
||||
extraHeaders['X-Proxy-Request-Id'] = proxyContext.requestId;
|
||||
}
|
||||
if (proxyContext?.cacheStatus) {
|
||||
extraHeaders['X-Proxy-Cache-Status'] = proxyContext.cacheStatus;
|
||||
}
|
||||
|
||||
if (!res.raw || !isFunction(res.raw.writeHead) || !isFunction(res.raw.write)) {
|
||||
res.header('Content-Type', 'text/event-stream');
|
||||
res.header('Cache-Control', 'no-cache');
|
||||
res.header('Connection', 'keep-alive');
|
||||
for (const [key, value] of Object.entries(extraHeaders)) {
|
||||
res.header(key, value);
|
||||
}
|
||||
res.send(stream);
|
||||
this.proxyMetrics?.decrementActiveConnections();
|
||||
return;
|
||||
@@ -1063,6 +1395,7 @@ export class ProxyController {
|
||||
'Content-Type': 'text/event-stream',
|
||||
'Cache-Control': 'no-cache',
|
||||
Connection: 'keep-alive',
|
||||
...extraHeaders,
|
||||
});
|
||||
|
||||
if (isFunction(res.raw.flushHeaders)) {
|
||||
|
||||
@@ -9,11 +9,24 @@ import { TokenUsageService } from '../usage/token-usage.service';
|
||||
import { ProxyReplayService } from './proxy-replay.service';
|
||||
import { ProxyMetricsService } from './proxy-metrics.service';
|
||||
import { ProxyIdeConfigService } from './proxy-ide-config.service';
|
||||
import { ProxyEventBus } from './proxy-event-bus.service';
|
||||
import { ProxyTransformInterceptor } from './proxy-transform.interceptor';
|
||||
|
||||
@Module({
|
||||
imports: [],
|
||||
controllers: [ProxyController, GeminiController],
|
||||
providers: [ProxyService, TokenManagerService, GeminiClient, ProxyGuard, TokenUsageService, ProxyReplayService, ProxyMetricsService, ProxyIdeConfigService],
|
||||
providers: [
|
||||
ProxyService,
|
||||
TokenManagerService,
|
||||
GeminiClient,
|
||||
ProxyGuard,
|
||||
TokenUsageService,
|
||||
ProxyReplayService,
|
||||
ProxyMetricsService,
|
||||
ProxyIdeConfigService,
|
||||
ProxyEventBus,
|
||||
ProxyTransformInterceptor,
|
||||
],
|
||||
exports: [TokenManagerService],
|
||||
})
|
||||
export class ProxyModule {}
|
||||
@@ -37,9 +37,26 @@ import { UpstreamRequestError } from './clients/upstream-error';
|
||||
import { trafficLogger } from '../../../utils/traffic-logger';
|
||||
import { encode } from 'gpt-tokenizer';
|
||||
|
||||
export interface ProxyRequestContext {
|
||||
requestId?: string;
|
||||
accountId?: string;
|
||||
retryCount?: number;
|
||||
model?: string;
|
||||
cacheStatus?: 'HIT' | 'MISS';
|
||||
}
|
||||
|
||||
export interface ProxyConfigOverride {
|
||||
accountId?: string;
|
||||
schedulingMode?: 'cache-first' | 'balance' | 'performance-first';
|
||||
maxRetries?: number;
|
||||
timeout?: number;
|
||||
}
|
||||
|
||||
@Injectable()
|
||||
export class ProxyService {
|
||||
private readonly logger = new Logger(ProxyService.name);
|
||||
private readonly responseCache = new Map<string, { response: OpenAIChatResponse; expiry: number }>();
|
||||
private readonly CACHE_TTL_MS = 60000; // 1 minute
|
||||
|
||||
constructor(
|
||||
@Inject(TokenManagerService) private readonly tokenManager: TokenManagerService,
|
||||
@@ -99,6 +116,8 @@ export class ProxyService {
|
||||
|
||||
async handleAnthropicMessages(
|
||||
request: AnthropicChatRequest,
|
||||
context?: ProxyRequestContext,
|
||||
configOverride?: ProxyConfigOverride,
|
||||
): Promise<AnthropicChatResponse | Observable<string>> {
|
||||
const sessionKey = this.extractAnthropicSessionKey(request);
|
||||
|
||||
@@ -124,11 +143,17 @@ export class ProxyService {
|
||||
sessionKey,
|
||||
excludeAccountIds: Array.from(attemptedAccountIds),
|
||||
model: targetModel,
|
||||
accountId: configOverride?.accountId,
|
||||
});
|
||||
if (!token) {
|
||||
throw new Error('No available accounts');
|
||||
}
|
||||
attemptedAccountIds.add(token.id);
|
||||
if (context) {
|
||||
context.accountId = token.id;
|
||||
context.retryCount = i;
|
||||
context.model = targetModel;
|
||||
}
|
||||
const effectiveTargetModel = this.tokenManager.resolveDynamicModelForAccount(
|
||||
token.id,
|
||||
targetModel,
|
||||
@@ -821,23 +846,35 @@ export class ProxyService {
|
||||
|
||||
async handleChatCompletions(
|
||||
request: OpenAIChatRequest,
|
||||
requestId?: string,
|
||||
headerSessionKey?: string,
|
||||
context?: ProxyRequestContext,
|
||||
configOverride?: ProxyConfigOverride,
|
||||
): Promise<OpenAIChatResponse | Observable<string>> {
|
||||
// Prefer the x-session-affinity header value (sent by OpenCode) over body-level session ids.
|
||||
const sessionKey =
|
||||
(headerSessionKey ? `openai:${headerSessionKey}` : undefined) ??
|
||||
this.extractOpenAISessionKey(request);
|
||||
|
||||
const targetModel = this.resolveTargetModel(request.model);
|
||||
const extraHeaders = this.createModelSpecificHeaders(request.model);
|
||||
this.logger.log(
|
||||
`OpenAI-compatible request received: model=${request.model}, mappedModel=${targetModel}, stream=${request.stream}`,
|
||||
`Anthropic request received: model=${request.model}, mappedModel=${targetModel}, stream=${request.stream}`,
|
||||
);
|
||||
const sessionKey = this.extractOpenAISessionKey(request);
|
||||
const requestId = uuidv4();
|
||||
|
||||
// Retry loop for account selection
|
||||
// Cache lookup for non-streaming requests
|
||||
if (!request.stream) {
|
||||
const cacheKey = this.buildCacheKey(request);
|
||||
const cached = this.getCachedResponse(cacheKey);
|
||||
if (cached) {
|
||||
if (context) {
|
||||
context.cacheStatus = 'HIT';
|
||||
context.accountId = 'cache';
|
||||
context.model = request.model;
|
||||
context.retryCount = 0;
|
||||
}
|
||||
return cached;
|
||||
}
|
||||
}
|
||||
|
||||
// Retry loop
|
||||
let lastError: unknown = null;
|
||||
const maxRetries = 3;
|
||||
const maxRetries = configOverride?.maxRetries ?? 3;
|
||||
const attemptedAccountIds = new Set<string>();
|
||||
|
||||
for (let i = 0; i < maxRetries; i++) {
|
||||
@@ -863,6 +900,12 @@ export class ProxyService {
|
||||
token.id,
|
||||
targetModel,
|
||||
);
|
||||
if (context) {
|
||||
context.accountId = token.id;
|
||||
context.retryCount = i;
|
||||
context.model = effectiveTargetModel;
|
||||
context.requestId = requestId;
|
||||
}
|
||||
|
||||
try {
|
||||
const claudeRequest = this.convertOpenAIToClaude(request);
|
||||
@@ -949,7 +992,13 @@ export class ProxyService {
|
||||
this.logger.log(
|
||||
`Transformed Claude response snippet: ${JSON.stringify(claudeResponse).substring(0, 500)}`,
|
||||
);
|
||||
return this.convertClaudeToOpenAIResponse(claudeResponse, request.model);
|
||||
const openaiResponse = this.convertClaudeToOpenAIResponse(claudeResponse, request.model);
|
||||
const transformedResponse = this.applyResponseFormatTransformation(openaiResponse, request);
|
||||
this.setCachedResponse(this.buildCacheKey(request), transformedResponse);
|
||||
if (context) {
|
||||
context.cacheStatus = 'MISS';
|
||||
}
|
||||
return transformedResponse;
|
||||
}
|
||||
} catch (err) {
|
||||
if (err instanceof Error && this.isProjectContextError(err.message)) {
|
||||
@@ -1007,7 +1056,13 @@ export class ProxyService {
|
||||
);
|
||||
this.recordUsage(token.id, effectiveTargetModel, response.usageMetadata, 'openai');
|
||||
const claudeResponse = transformResponse(response);
|
||||
return this.convertClaudeToOpenAIResponse(claudeResponse, request.model);
|
||||
const openaiResponse = this.convertClaudeToOpenAIResponse(claudeResponse, request.model);
|
||||
const transformedResponse = this.applyResponseFormatTransformation(openaiResponse, request);
|
||||
this.setCachedResponse(this.buildCacheKey(request), transformedResponse);
|
||||
if (context) {
|
||||
context.cacheStatus = 'MISS';
|
||||
}
|
||||
return transformedResponse;
|
||||
} catch (fallbackErr) {
|
||||
lastError = fallbackErr;
|
||||
}
|
||||
@@ -1593,6 +1648,10 @@ export class ProxyService {
|
||||
});
|
||||
}
|
||||
|
||||
if (request.response_format?.type === 'json_object' || request.response_format?.type === 'json_schema') {
|
||||
systemPromptParts.push('You must format your entire response as a single valid JSON object. Do not include any markdown formatting, explanations, or text outside the JSON structure.');
|
||||
}
|
||||
|
||||
const systemPrompt = systemPromptParts.length > 0 ? systemPromptParts.join('\n') : undefined;
|
||||
|
||||
return {
|
||||
@@ -2073,6 +2132,75 @@ export class ProxyService {
|
||||
);
|
||||
}
|
||||
|
||||
private applyResponseFormatTransformation(
|
||||
response: OpenAIChatResponse,
|
||||
request: OpenAIChatRequest,
|
||||
): OpenAIChatResponse {
|
||||
if (!request.response_format?.type) {
|
||||
return response;
|
||||
}
|
||||
const type = request.response_format.type;
|
||||
if (type !== 'json_object' && type !== 'json_schema') {
|
||||
return response;
|
||||
}
|
||||
|
||||
const content = response.choices?.[0]?.message?.content;
|
||||
if (!isString(content)) {
|
||||
return response;
|
||||
}
|
||||
|
||||
// Try to extract JSON from markdown code blocks
|
||||
let jsonText = content;
|
||||
const codeBlockMatch = content.match(/```(?:json)?\s*([\s\S]*?)\s*```/);
|
||||
if (codeBlockMatch?.[1]) {
|
||||
jsonText = codeBlockMatch[1].trim();
|
||||
}
|
||||
|
||||
try {
|
||||
// Validate it's parseable JSON
|
||||
JSON.parse(jsonText);
|
||||
// Replace the content with the extracted JSON string
|
||||
const newResponse = { ...response };
|
||||
newResponse.choices = [...response.choices];
|
||||
newResponse.choices[0] = { ...response.choices[0] };
|
||||
newResponse.choices[0].message = { ...response.choices[0].message };
|
||||
newResponse.choices[0].message.content = jsonText;
|
||||
return newResponse;
|
||||
} catch {
|
||||
// If not valid JSON, return original unchanged
|
||||
return response;
|
||||
}
|
||||
}
|
||||
|
||||
private buildCacheKey(request: OpenAIChatRequest): string {
|
||||
const keyBody = {
|
||||
model: request.model,
|
||||
messages: request.messages,
|
||||
temperature: request.temperature,
|
||||
top_p: request.top_p,
|
||||
max_tokens: request.max_tokens,
|
||||
tools: request.tools,
|
||||
response_format: request.response_format,
|
||||
};
|
||||
return JSON.stringify(keyBody);
|
||||
}
|
||||
|
||||
private getCachedResponse(key: string): OpenAIChatResponse | undefined {
|
||||
const entry = this.responseCache.get(key);
|
||||
if (!entry) {
|
||||
return undefined;
|
||||
}
|
||||
if (Date.now() > entry.expiry) {
|
||||
this.responseCache.delete(key);
|
||||
return undefined;
|
||||
}
|
||||
return entry.response;
|
||||
}
|
||||
|
||||
private setCachedResponse(key: string, response: OpenAIChatResponse): void {
|
||||
this.responseCache.set(key, { response, expiry: Date.now() + this.CACHE_TTL_MS });
|
||||
}
|
||||
|
||||
private extractAnthropicSessionKey(request: AnthropicChatRequest): string | undefined {
|
||||
const metadata = request.metadata;
|
||||
const sessionCandidate =
|
||||
|
||||
@@ -33,6 +33,7 @@ interface GetNextTokenOptions {
|
||||
sessionKey?: string;
|
||||
excludeAccountIds?: string[];
|
||||
model?: string;
|
||||
accountId?: string;
|
||||
}
|
||||
|
||||
type TokenEntry = [string, TokenData];
|
||||
@@ -339,10 +340,26 @@ export class TokenManagerService implements OnModuleInit {
|
||||
const sessionKey = options?.sessionKey?.trim();
|
||||
const model = options?.model;
|
||||
const excludedAccountIds = new Set(options?.excludeAccountIds ?? []);
|
||||
const requestedAccountId = options?.accountId?.trim();
|
||||
|
||||
this.clearExpiredSessionBindings(now);
|
||||
this.rateLimitTracker.cleanupExpired();
|
||||
|
||||
// If a specific account is requested, try to use it first
|
||||
if (requestedAccountId && !excludedAccountIds.has(requestedAccountId)) {
|
||||
const requestedToken = this.tokens.get(requestedAccountId);
|
||||
if (requestedToken) {
|
||||
const isHealthy = this.isAccountHealthy(requestedAccountId, requestedToken, nowSeconds, model);
|
||||
if (isHealthy) {
|
||||
this.logger.log(`Using requested account: ${requestedAccountId}`);
|
||||
return this.buildCloudAccount(requestedAccountId, requestedToken);
|
||||
}
|
||||
this.logger.warn(`Requested account ${requestedAccountId} is not healthy, falling back to selection`);
|
||||
} else {
|
||||
this.logger.warn(`Requested account ${requestedAccountId} not found, falling back to selection`);
|
||||
}
|
||||
}
|
||||
|
||||
const fullAccountPool = Array.from(this.tokens.entries());
|
||||
const filteredAccountPool = fullAccountPool.filter(
|
||||
([accountId]) => !excludedAccountIds.has(accountId),
|
||||
@@ -945,6 +962,42 @@ export class TokenManagerService implements OnModuleInit {
|
||||
}
|
||||
}
|
||||
|
||||
private isAccountHealthy(accountId: string, tokenData: TokenData, nowSeconds: number, model?: string): boolean {
|
||||
const entry = this.getOrCreateCircuitBreakerEntry(accountId);
|
||||
if (entry.state === 'unhealthy') {
|
||||
return false;
|
||||
}
|
||||
if (this.rateLimitTracker.isRateLimited(accountId, model)) {
|
||||
return false;
|
||||
}
|
||||
if (nowSeconds >= tokenData.expiry_timestamp) {
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
private buildCloudAccount(accountId: string, tokenData: TokenData): CloudAccount {
|
||||
const timestamp = Date.now();
|
||||
return {
|
||||
id: accountId,
|
||||
provider: 'google',
|
||||
email: tokenData.email,
|
||||
token: {
|
||||
access_token: tokenData.access_token,
|
||||
refresh_token: tokenData.refresh_token,
|
||||
token_type: tokenData.token_type,
|
||||
expires_in: tokenData.expires_in,
|
||||
expiry_timestamp: tokenData.expiry_timestamp,
|
||||
project_id: tokenData.project_id,
|
||||
oauth_client_key: tokenData.oauth_client_key,
|
||||
session_id: tokenData.session_id,
|
||||
upstream_proxy_url: tokenData.upstream_proxy_url,
|
||||
},
|
||||
created_at: timestamp,
|
||||
last_used: timestamp,
|
||||
};
|
||||
}
|
||||
|
||||
private resolveAccountId(accountIdOrEmail: string): string | null {
|
||||
if (this.tokens.has(accountIdOrEmail)) {
|
||||
return accountIdOrEmail;
|
||||
|
||||
Reference in new issue
Block a user