From 42436bca531060d6d4effff93706c6af3f9e65b5 Mon Sep 17 00:00:00 2001 From: RainbowBird Date: Sat, 30 May 2026 20:13:53 +0800 Subject: [PATCH] refactor(server): langfuse upstream model --- .../verifications/langfuse-tracing.md | 18 ++++++ apps/server/src/routes/openai/v1/index.ts | 38 +++++++----- .../server/src/routes/openai/v1/route.test.ts | 60 ++++++++++++++++++- .../src/services/domain/llm-router/router.ts | 9 ++- .../domain/llm-router/tests/router.test.ts | 10 ++-- .../src/services/domain/llm-router/types.ts | 2 + 6 files changed, 114 insertions(+), 23 deletions(-) diff --git a/apps/server/docs/ai-context/verifications/langfuse-tracing.md b/apps/server/docs/ai-context/verifications/langfuse-tracing.md index 9bd32a35a..40937002a 100644 --- a/apps/server/docs/ai-context/verifications/langfuse-tracing.md +++ b/apps/server/docs/ai-context/verifications/langfuse-tracing.md @@ -112,6 +112,24 @@ Langfuse 隔离 live smoke(覆盖 `NODE_ENV=codex-langfuse-smoke`, `OTEL_SERVICE 仍未做真实 server HTTP E2E:本地 `.env.local` 里的 DB/Redis 仍指向生产实例。当前已验证的是同一 `instrumentation.ts` + `llm-tracing` generation SDK 写入链路;真实 HTTP 请求还需要 staging DB/Redis/router/auth token 后补跑。 +## 模型归因修正(chat-auto alias → 上游模型) + +Langfuse Model costs 页面曾出现 `chat-auto`。这不是 Langfuse pricing 配置问题,而是 route 在调用 `llmRouter.route(...)` 前就用 client/request model 创建 `chat.completion` generation;如果 router config 通过 `upstream.overrideModel` 把 `chat-auto` 改写成真实上游模型,Langfuse 仍记录 alias。 + +修正: + +- `LlmRouteContext.upstreamModel`:router 成功命中上游时写入实际发给上游的 `overrideModel ?? modelName`。 +- `handleCompletion`:router 返回后再创建 Langfuse generation,`model` 使用 `routeCtx.upstreamModel ?? requestModel`。 +- route 里的 billing/request-log/本地 OTel metric 仍保持原有 `requestModel` 语义;本次只修 Langfuse model-cost 归因。 + +复测: + +- `apps/server/src/services/domain/llm-router/tests/router.test.ts`:覆盖 `upstream.overrideModel` 同时写入 `ctx.upstreamModel`。 +- `apps/server/src/routes/openai/v1/route.test.ts`:覆盖请求 `model=chat-auto`、router context 返回 `openai/gpt-4o-mini` 时,`startChatGeneration({ model })` 使用 `openai/gpt-4o-mini`。 +- `pnpm exec vitest run apps/server/src/services/domain/llm-router/tests/router.test.ts apps/server/src/routes/openai/v1/route.test.ts apps/server/src/services/domain/llm-tracing/index.test.ts`:3 files / 78 tests passed。 +- `pnpm -F @proj-airi/server typecheck`:0 错误。 +- `pnpm exec eslint apps/server/src/routes/openai/v1/index.ts apps/server/src/routes/openai/v1/route.test.ts apps/server/src/services/domain/llm-router/router.ts apps/server/src/services/domain/llm-router/types.ts apps/server/src/services/domain/llm-router/tests/router.test.ts`:0 输出。 + ## 环境 - base commit: `dc1037f34`(本次改动未提交,工作树状态) diff --git a/apps/server/src/routes/openai/v1/index.ts b/apps/server/src/routes/openai/v1/index.ts index bfc04c745..90841aff0 100644 --- a/apps/server/src/routes/openai/v1/index.ts +++ b/apps/server/src/routes/openai/v1/index.ts @@ -178,20 +178,6 @@ export function createV1Routes( }, }) - // Langfuse LLM-native generation: per-request prompt/completion record - // (input/output/model/usage) powering prompt trace, eval, and per-user/ - // session cost. The llm-tracing module hides the enable gate, SDK shape, SSE - // assembly, and lifecycle — this is a no-op handle when tracing is off, and - // succeed/fail are idempotent, so every exit branch below can close it. - const generationTrace = startChatGeneration({ - input: body.messages, - model: requestModel, - requestId, - stream, - userId: user.id, - sessionId: c.req.header('x-airi-session-id'), - }) - const startedAt = Date.now() // Router throws ApiError (502/503/504/400) on full exhaustion or unknown @@ -213,13 +199,35 @@ export function createV1Routes( catch (err) { span.setStatus({ code: SpanStatusCode.ERROR, message: 'Router exhausted or unknown model' }) span.end() - generationTrace.fail('Router exhausted or unknown model') + startChatGeneration({ + input: body.messages, + model: routeCtx.upstreamModel ?? requestModel, + requestId, + stream, + userId: user.id, + sessionId: c.req.header('x-airi-session-id'), + }).fail('Router exhausted or unknown model') recordMetrics({ model: requestModel, status: 502, type: 'chat', provider: routeCtx.provider, durationMs: Date.now() - startedAt, fluxConsumed: 0 }) throw err } const durationMs = Date.now() - startedAt span.setAttribute('http.response.status_code', response.status) + const langfuseModel = routeCtx.upstreamModel ?? requestModel + + // Langfuse LLM-native generation: per-request prompt/completion record + // (input/output/model/usage) powering prompt trace, eval, and per-user/ + // session cost. Use the router-resolved upstream model, not the client + // alias (`auto` / `chat-auto`), so Langfuse model-cost grouping matches the + // provider model that actually generated the tokens. + const generationTrace = startChatGeneration({ + input: body.messages, + model: langfuseModel, + requestId, + stream, + userId: user.id, + sessionId: c.req.header('x-airi-session-id'), + }) if (!response.ok) { span.setStatus({ code: SpanStatusCode.ERROR, message: `Gateway ${response.status}` }) diff --git a/apps/server/src/routes/openai/v1/route.test.ts b/apps/server/src/routes/openai/v1/route.test.ts index a34a58cf1..aba38b2cb 100644 --- a/apps/server/src/routes/openai/v1/route.test.ts +++ b/apps/server/src/routes/openai/v1/route.test.ts @@ -6,11 +6,28 @@ import type { RequestLogService } from '../../../services/domain/request-log' import type { HonoEnv } from '../../../types/hono' import { Hono } from 'hono' -import { afterAll, describe, expect, it, vi } from 'vitest' +import { afterAll, beforeEach, describe, expect, it, vi } from 'vitest' import { createV1Routes } from '.' import { ApiError } from '../../../utils/error' +const tracingSpies = vi.hoisted(() => ({ + startChatGeneration: vi.fn(() => ({ + appendStreamChunk: vi.fn(), + succeed: vi.fn(), + fail: vi.fn(), + })), + startTtsGeneration: vi.fn(() => ({ + succeed: vi.fn(), + fail: vi.fn(), + })), +})) + +vi.mock('../../../services/domain/llm-tracing', () => ({ + startChatGeneration: tracingSpies.startChatGeneration, + startTtsGeneration: tracingSpies.startTtsGeneration, +})) + function createMockFluxService(flux = 100): FluxService { return { getFlux: vi.fn(async () => ({ userId: 'user-1', flux })), @@ -166,6 +183,11 @@ const testUser = { id: 'user-1', name: 'Test User', email: 'test@example.com' } describe('v1CompletionsRoutes', () => { const originalFetch = globalThis.fetch + beforeEach(() => { + tracingSpies.startChatGeneration.mockClear() + tracingSpies.startTtsGeneration.mockClear() + }) + afterAll(() => { globalThis.fetch = originalFetch }) @@ -381,6 +403,42 @@ describe('v1CompletionsRoutes', () => { ) }) + it('records Langfuse chat generation with the router-resolved upstream model', async () => { + const llmRouter = createMockLlmRouter({ + route: vi.fn(async (_req, ctx) => { + if (ctx) { + ctx.provider = 'openrouter' + ctx.upstreamModel = 'openai/gpt-4o-mini' + } + return new Response(JSON.stringify({ + choices: [], + usage: { prompt_tokens: 1, completion_tokens: 2, total_tokens: 3 }, + }), { + status: 200, + headers: { 'Content-Type': 'application/json' }, + }) + }) as any, + }) + const app = createTestApp(createMockFluxService(), createMockConfigKV(), undefined, undefined, undefined, llmRouter) + + await app.fetch( + new Request('http://localhost/api/v1/openai/chat/completions', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ model: 'chat-auto', messages: [{ role: 'user', content: 'hi' }] }), + }), + { user: testUser } as any, + ) + + expect(tracingSpies.startChatGeneration).toHaveBeenCalledWith( + expect.objectContaining({ + model: 'openai/gpt-4o-mini', + requestId: expect.any(String), + userId: 'user-1', + }), + ) + }) + it('should not charge flux when upstream returns error', async () => { globalThis.fetch = vi.fn(async () => new Response('{"error":"bad"}', { status: 500, diff --git a/apps/server/src/services/domain/llm-router/router.ts b/apps/server/src/services/domain/llm-router/router.ts index 93a7661b8..8112f4d8b 100644 --- a/apps/server/src/services/domain/llm-router/router.ts +++ b/apps/server/src/services/domain/llm-router/router.ts @@ -195,7 +195,7 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { fallbackHttpCodes: number[], onAttemptFailure: (failure: { keyId: string, status: number | 'timeout', bodySnippet?: string, errorMessage?: string }) => void, ): Promise< - | { kind: 'ok', response: Response, attemptIndex: number } + | { kind: 'ok', response: Response, attemptIndex: number, upstreamModel: string } | { kind: 'exhausted', failures: Array<{ keyId: string, status: number | 'timeout', bodySnippet?: string, errorMessage?: string }> } > { const provider = deriveProviderTag(upstream.baseURL) @@ -256,7 +256,7 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { [AIRI_ATTR_GEN_AI_GATEWAY_KEY_ID]: key.id, [AIRI_ATTR_GEN_AI_GATEWAY_FALLBACK_DEPTH]: attemptIndex, }) - return { kind: 'ok', response, attemptIndex } + return { kind: 'ok', response, attemptIndex, upstreamModel: effectiveModel } } const status = response.status @@ -365,8 +365,11 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { (failure) => { allFailures.push({ provider, ...failure }) }, ) - if (result.kind === 'ok') + if (result.kind === 'ok') { + if (ctx) + ctx.upstreamModel = result.upstreamModel return result.response + } // This upstream exhausted; record and continue. options.gatewayMetrics?.keyExhaustedCount.add(1, { provider }) diff --git a/apps/server/src/services/domain/llm-router/tests/router.test.ts b/apps/server/src/services/domain/llm-router/tests/router.test.ts index 95b4a9f6f..df4f89a47 100644 --- a/apps/server/src/services/domain/llm-router/tests/router.test.ts +++ b/apps/server/src/services/domain/llm-router/tests/router.test.ts @@ -5,7 +5,7 @@ import type Redis from 'ioredis' import type { GatewayMetrics } from '../../../../otel' import type { ConfigKVService } from '../../../adapters/config-kv' -import type { RouterConfig } from '../types' +import type { LlmRouteContext, RouterConfig } from '../types' import { randomBytes } from 'node:crypto' @@ -174,7 +174,7 @@ describe('createLlmRouterService', () => { redis: makeRedisStub(), }) - const ctx = { provider: 'unknown', triedUpstreams: 0, triedKeys: 0, lastStatus: null } + const ctx: LlmRouteContext = { provider: 'unknown', triedUpstreams: 0, triedKeys: 0, lastStatus: null } const res = await router.route({ modelName: 'openai/gpt-5-mini', body: { messages: [] } }, ctx) expect(res.status).toBe(200) // deriveProviderTag = URL hostname. @@ -205,7 +205,7 @@ describe('createLlmRouterService', () => { redis: makeRedisStub(), }) - const ctx = { provider: 'unknown', triedUpstreams: 0, triedKeys: 0, lastStatus: null } + const ctx: LlmRouteContext = { provider: 'unknown', triedUpstreams: 0, triedKeys: 0, lastStatus: null } const res = await router.route({ modelName: 'openai/gpt-5-mini', body: {} }, ctx) expect(res.status).toBe(200) expect(ctx.provider).toBe('up-b.example') @@ -248,10 +248,12 @@ describe('createLlmRouterService', () => { redis: makeRedisStub(), }) - await router.route({ modelName: 'openai/gpt-5-mini', body: { messages: [] } }) + const ctx: LlmRouteContext = { provider: 'unknown', triedUpstreams: 0, triedKeys: 0, lastStatus: null } + await router.route({ modelName: 'openai/gpt-5-mini', body: { messages: [] } }, ctx) const calls = (fetchImpl as unknown as ReturnType).mock.calls const init = calls[0][1] as { body: string } expect((JSON.parse(init.body) as { model: string }).model).toBe('real/upstream-id') + expect(ctx.upstreamModel).toBe('real/upstream-id') }) it('multi-key fallback: k1=401 then k2=200 → returns 200 and records fallbackCount once', async () => { diff --git a/apps/server/src/services/domain/llm-router/types.ts b/apps/server/src/services/domain/llm-router/types.ts index 9a477130f..5e8144cd9 100644 --- a/apps/server/src/services/domain/llm-router/types.ts +++ b/apps/server/src/services/domain/llm-router/types.ts @@ -104,6 +104,8 @@ export interface LlmRouteRequest { export interface LlmRouteContext { /** Provider tag for OTel labels (e.g. `openrouter`). */ provider: string + /** Actual model id sent to the winning upstream after `overrideModel` rewrites. */ + upstreamModel?: string /** Number of upstreams attempted so far. */ triedUpstreams: number /** Number of keys attempted across all upstreams so far. */