// otel-init.ts —— 必须在业务代码之前 import
import { NodeSDK } from '@opentelemetry/sdk-node';
import { getNodeAutoInstrumentations } from '@opentelemetry/auto-instrumentations-node';
import { OTLPTraceExporter } from '@opentelemetry/exporter-trace-otlp-http';
import { Resource } from '@opentelemetry/resources';
import { ATTR_SERVICE_NAME, ATTR_SERVICE_VERSION } from '@opentelemetry/semantic-conventions';
 
const sdk = new NodeSDK({
  resource: new Resource({
    [ATTR_SERVICE_NAME]: 'agentflow-api',
    [ATTR_SERVICE_VERSION]: process.env.SERVICE_VERSION ?? 'dev',
    'deployment.environment': process.env.NODE_ENV ?? 'development',
  }),
  traceExporter: new OTLPTraceExporter({
    url: process.env.OTEL_EXPORTER_OTLP_ENDPOINT ?? 'http://localhost:4318/v1/traces',
  }),
  instrumentations: [getNodeAutoInstrumentations()],
});
 
sdk.start();

// src/index.ts
import './otel-init';   // 必须放第一行
import { startServer } from './server';
 
startServer();

import { trace, SpanStatusCode, SpanKind } from '@opentelemetry/api';
 
const tracer = trace.getTracer('agent-engine', '1.0.0');
 
export async function runAgentStep(input: AgentInput): Promise<AgentOutput> {
  // 一次 Agent step 是 trace 里的根 span（如果是 HTTP 触发的，则是子 span）
  return tracer.startActiveSpan(
    'agent.step',
    { kind: SpanKind.INTERNAL },
    async (span) => {
      // 业务标签全部用低基数维度
      span.setAttribute('tenant.id', input.tenantId);
      span.setAttribute('session.id', input.sessionId);
      span.setAttribute('agent.step.type', input.stepType);
      span.setAttribute('agent.step.iteration', input.iteration);
 
      try {
        // Planner：决定下一步做什么
        const plan = await tracer.startActiveSpan('agent.planner', async (s) => {
          // GenAI semantic conventions —— OTel 的半官方标准，下一节详述
          s.setAttribute('gen_ai.system', 'anthropic');
          s.setAttribute('gen_ai.request.model', 'claude-sonnet-4-5');
          s.setAttribute('gen_ai.request.temperature', 0.2);
 
          const result = await llmGateway.complete({
            model: 'claude-sonnet-4-5',
            messages: input.messages,
            tenantId: input.tenantId,
          });
 
          s.setAttribute('gen_ai.usage.input_tokens', result.usage.inputTokens);
          s.setAttribute('gen_ai.usage.output_tokens', result.usage.outputTokens);
          s.setAttribute('gen_ai.usage.cache_read_input_tokens', result.usage.cacheReadTokens ?? 0);
          s.setAttribute('gen_ai.response.id', result.id);
          s.end();
          return result;
        });
 
        // Executor：并行调用多个 Skill
        const skillResults = await tracer.startActiveSpan('agent.executor', async (s) => {
          s.setAttribute('agent.executor.skill_count', plan.toolCalls.length);
          const results = await Promise.all(
            plan.toolCalls.map((call) =>
              tracer.startActiveSpan(`skill.${call.name}`, async (cs) => {
                cs.setAttribute('skill.name', call.name);
                cs.setAttribute('skill.input.size', JSON.stringify(call.args).length);
                try {
                  const r = await skillRegistry.invoke(call.name, call.args);
                  cs.setAttribute('skill.output.size', JSON.stringify(r).length);
                  return r;
                } finally {
                  cs.end();
                }
              }),
            ),
          );
          s.end();
          return results;
        });
 
        // Reflector：决定是否继续下一步
        // ……（略）
 
        span.setStatus({ code: SpanStatusCode.OK });
        return { plan, skillResults };
      } catch (err) {
        span.recordException(err as Error);
        span.setStatus({ code: SpanStatusCode.ERROR, message: (err as Error).message });
        throw err;
      } finally {
        span.end();
      }
    },
  );
}

import { trace, propagation, context, SpanKind } from '@opentelemetry/api';
import { Queue, Worker } from 'bullmq';
 
const tracer = trace.getTracer('agent-task');
 
// ============ 生产端：投任务时注入 context ============
export async function enqueueAgentTask(queue: Queue, data: TaskPayload): Promise<void> {
  // 用一个普通对象当 carrier，propagation 会把当前 active context 序列化进去
  const carrier: Record<string, string> = {};
  propagation.inject(context.active(), carrier);
 
  await queue.add('agent-task', {
    ...data,
    _otelCarrier: carrier, // 跟随 job data 一起持久化到 Redis
  });
}
 
// ============ 消费端：处理任务前提取 context ============
export function createAgentWorker(queueName: string) {
  return new Worker(
    queueName,
    async (job) => {
      const carrier = (job.data._otelCarrier ?? {}) as Record<string, string>;
      // 把 carrier 里的 traceparent 提取回 OTel context
      const parentCtx = propagation.extract(context.active(), carrier);
 
      // 用 SpanKind.CONSUMER 表示这是消息队列的消费者侧
      return context.with(parentCtx, async () => {
        return tracer.startActiveSpan(
          'agent-task.process',
          { kind: SpanKind.CONSUMER },
          async (span) => {
            span.setAttribute('job.id', job.id ?? 'unknown');
            span.setAttribute('queue.name', queueName);
            try {
              const result = await processAgentTask(job.data);
              return result;
            } catch (err) {
              span.recordException(err as Error);
              throw err;
            } finally {
              span.end();
            }
          },
        );
      });
    },
    { connection: { host: 'localhost', port: 6379 } },
  );
}

import { Langfuse } from 'langfuse';
 
const langfuse = new Langfuse({
  publicKey: process.env.LANGFUSE_PUBLIC_KEY!,
  secretKey: process.env.LANGFUSE_SECRET_KEY!,
  baseUrl: process.env.LANGFUSE_BASE_URL ?? 'http://localhost:3000',
  // 默认 batch flush，关进程前要 await shutdownAsync 否则丢数据
  flushAt: 15,
  flushInterval: 10_000,
});
 
export async function chatWithLangfuse(input: ChatInput): Promise<ChatOutput> {
  // 创建一个 trace —— 对应一次完整对话
  const lfTrace = langfuse.trace({
    name: 'customer-service-chat',
    userId: input.userId,
    sessionId: input.sessionId,
    metadata: {
      tenantId: input.tenantId,
      requestId: input.requestId,
    },
    tags: ['production', input.tenantId],
  });
 
  // 在 trace 下创建一个 generation —— 对应一次 LLM 调用
  const generation = lfTrace.generation({
    name: 'main-llm-call',
    model: 'claude-sonnet-4-5',
    modelParameters: { temperature: 0.2, maxTokens: 2048 },
    input: input.messages,
  });
 
  try {
    const response = await llmGateway.complete({
      messages: input.messages,
      model: 'claude-sonnet-4-5',
    });
 
    // 结束 generation —— 关键：要带 usage，Langfuse 据此算成本
    generation.end({
      output: response.content,
      usage: {
        input: response.usage.inputTokens,
        output: response.usage.outputTokens,
        total: response.usage.totalTokens,
        unit: 'TOKENS',
      },
    });
 
    lfTrace.update({ output: response.content });
    return { content: response.content };
  } catch (err) {
    generation.end({ level: 'ERROR', statusMessage: (err as Error).message });
    throw err;
  }
}

fastify.addHook('onClose', async () => {
  await langfuse.shutdownAsync();
});

// otel-init.ts
import { BatchSpanProcessor } from '@opentelemetry/sdk-trace-base';
import { OTLPTraceExporter } from '@opentelemetry/exporter-trace-otlp-http';
 
const jaegerExporter = new OTLPTraceExporter({
  url: 'http://localhost:4318/v1/traces',
});
const langfuseExporter = new OTLPTraceExporter({
  url: 'http://localhost:3000/api/public/otel/v1/traces',
  headers: {
    Authorization: `Basic ${Buffer.from(
      `${process.env.LANGFUSE_PUBLIC_KEY}:${process.env.LANGFUSE_SECRET_KEY}`,
    ).toString('base64')}`,
  },
});
 
const sdk = new NodeSDK({
  // ...
  spanProcessors: [
    new BatchSpanProcessor(jaegerExporter),
    new BatchSpanProcessor(langfuseExporter),
  ],
});

import { Langfuse } from 'langfuse';
 
const langfuse = new Langfuse({ /* ... */ });
 
export async function runAgentWithVersionedPrompt(input: AgentInput) {
  // 拉取 production label 的最新版本，带 60 秒缓存
  const prompt = await langfuse.getPrompt(
    'customer-service-system',
    undefined,
    { cacheTtlSeconds: 60, label: 'production' },
  );
 
  // 编译模板变量（langfuse 支持 mustache 风格的 {{var}}）
  const compiled = prompt.compile({
    tenantName: input.tenantName,
    productName: input.productName,
    currentDate: new Date().toISOString(),
  });
 
  // 这次对话先开一个 trace，再挂 generation
  const lfTrace = langfuse.trace({
    name: 'customer-service-chat',
    sessionId: input.sessionId,
  });
 
  const generation = lfTrace.generation({
    name: 'agent-main',
    model: 'claude-sonnet-4-5',
    prompt, // 关联 prompt 版本 —— 关键
    input: [{ role: 'system', content: compiled }, ...input.messages],
  });
 
  const result = await llmGateway.complete({
    system: compiled,
    messages: input.messages,
  });
 
  generation.end({
    output: result.content,
    usage: { input: result.usage.inputTokens, output: result.usage.outputTokens, unit: 'TOKENS' },
  });
}

function pickPromptLabel(sessionId: string): 'production' | 'experiment' {
  const hash = createHash('sha1').update(sessionId).digest()[0];
  return hash < 13 ? 'experiment' : 'production'; // 5% 走 experiment
}

import { z } from 'zod';
 
const RelevanceScoreSchema = z.object({
  score: z.number().min(1).max(5),
  reasoning: z.string(),
});
 
export async function evaluateRelevance(
  userMessage: string,
  agentResponse: string,
): Promise<z.infer<typeof RelevanceScoreSchema>> {
  const judgePrompt = `你是评估专家。给定用户问题和 Agent 回复，评估回复的相关性。
 
评分规则：
- 5 = 完全相关，直接命中用户意图
- 4 = 相关，覆盖主要意图
- 3 = 部分相关，遗漏关键信息
- 2 = 弱相关，答非所问但有部分关联
- 1 = 完全不相关
 
用户问题：${userMessage}
Agent 回复：${agentResponse}
 
只输出严格的 JSON：{"score": number, "reasoning": "string"}`;
 
  const result = await llmGateway.complete({
    model: 'claude-sonnet-4-5',
    messages: [{ role: 'user', content: judgePrompt }],
    temperature: 0, // 评估场景 temperature 必须 = 0
  });
 
  return RelevanceScoreSchema.parse(JSON.parse(result.content));
}

const PairwiseSchema = z.object({
  winner: z.enum(['A', 'B', 'tie']),
  reasoning: z.string(),
});
 
export async function pairwiseCompare(
  userMessage: string,
  responseA: string,
  responseB: string,
) {
  // 位置偏见：LLM 倾向于选第一个，要随机交换位置后再判
  const swap = Math.random() < 0.5;
  const [first, second] = swap ? [responseB, responseA] : [responseA, responseB];
 
  const result = await llmGateway.complete({
    model: 'claude-sonnet-4-5',
    temperature: 0,
    messages: [
      {
        role: 'user',
        content: `给定用户问题和两个 Agent 回复，判断哪个更好。
 
用户问题：${userMessage}
 
回复 1：${first}
回复 2：${second}
 
只输出 JSON：{"winner": "1" | "2" | "tie", "reasoning": "..."}`,
      },
    ],
  });
 
  const raw = JSON.parse(result.content);
  // 翻译回原始 A/B
  let winner: 'A' | 'B' | 'tie';
  if (raw.winner === 'tie') winner = 'tie';
  else if (raw.winner === '1') winner = swap ? 'B' : 'A';
  else winner = swap ? 'A' : 'B';
 
  return { winner, reasoning: raw.reasoning };
}

const baseline = await loadBaselineMetrics(); // 上一次发版的评估结果
const current = await runEvalsOnDataset('regression-set');
 
if (current.relevance < baseline.relevance - 0.05) {
  throw new Error(`Relevance 退步 ${baseline.relevance - current.relevance}，阻断发布`);
}
if (current.cost > baseline.cost * 1.2) {
  throw new Error(`成本上涨超过 20%，阻断发布`);
}