Agent 可观测性、监控与 Tracing 体系

// Trace 数据结构
interface Trace {
  traceId: string;           // 128 位追踪 ID（32 字符十六进制）
  name: string;              // Trace 名称
  startTime: number;         // 开始时间（Unix 时间戳，纳秒）
  endTime: number;           // 结束时间
  spans: Span[];             // 包含的 Span 列表
  resource: Resource;        // 资源信息（服务名、版本等）
}

// Span 数据结构
interface Span {
  spanId: string;            // 64 位跨度 ID（16 字符十六进制）
  traceId: string;           // 所属 Trace ID
  parentSpanId?: string;     // 父 Span ID（根 Span 为空）
  name: string;              // Span 名称
  kind: SpanKind;            // Span 类型
  startTime: number;         // 开始时间（纳秒）
  endTime: number;           // 结束时间
  status: SpanStatus;        // 状态（OK/ERROR/UNSET）
  
  // 属性
  attributes: {              // Tags/Attributes
    [key: string]: string | number | boolean;
  };
  
  // 事件
  events: SpanEvent[];       // 时间线事件
  
  // 链接
  links: SpanLink[];         // 与其他 Trace 的链接
  
  // 资源
  resource: Resource;        // 资源信息
}

// Span 类型
enum SpanKind {
  INTERNAL = 0,              // 内部操作
  SERVER = 1,                // 服务端处理请求
  CLIENT = 2,                // 客户端发起请求
  PRODUCER = 3,              // 消息生产者
  CONSUMER = 4               // 消息消费者
}

// Span 状态
interface SpanStatus {
  code: StatusCode;          // OK | ERROR | UNSET
  message?: string;          // 错误描述
}

// Span 事件
interface SpanEvent {
  name: string;              // 事件名称
  timestamp: number;         // 事件时间（纳秒）
  attributes?: {             // 事件属性
    [key: string]: string | number | boolean;
  };
}

// Span 链接
interface SpanLink {
  traceId: string;           // 链接的 Trace ID
  spanId: string;            // 链接的 Span ID
  attributes?: {             // 链接属性
    [key: string]: string | number | boolean;
  };
}

// 资源信息
interface Resource {
  attributes: {
    'service.name': string;          // 服务名称
    'service.version': string;       // 服务版本
    'service.instance.id': string;   // 服务实例 ID
    'deployment.environment': string;// 部署环境
    [key: string]: string;
  };
}

// 示例：Agent 对话 Trace
const agentConversationTrace: Trace = {
  traceId: '4bf92f3577b34da6a3ce929d0e0e4736',
  name: 'customer-support-conversation',
  startTime: 1709990400000000000,
  endTime: 1709990402500000000,
  resource: {
    attributes: {
      'service.name': 'customer-support-agent',
      'service.version': '2.1.0',
      'deployment.environment': 'production'
    }
  },
  spans: [
    {
      spanId: '00f067aa0ba902b7',
      traceId: '4bf92f3577b34da6a3ce929d0e0e4736',
      name: 'receive-user-message',
      kind: SpanKind.SERVER,
      startTime: 1709990400000000000,
      endTime: 1709990400050000000,
      status: { code: 'OK' },
      attributes: {
        'user.id': 'user-12345',
        'message.length': 128
      }
    },
    {
      spanId: '6e0c63257de34c92',
      traceId: '4bf92f3577b34da6a3ce929d0e0e4736',
      parentSpanId: '00f067aa0ba902b7',
      name: 'llm-completion',
      kind: SpanKind.CLIENT,
      startTime: 1709990400100000000,
      endTime: 1709990401800000000,
      status: { code: 'OK' },
      attributes: {
        'llm.model': 'gpt-4-turbo',
        'llm.temperature': 0.7,
        'llm.token_count': 342
      }
    }
  ]
};
                        

// W3C Trace Context 标准
// traceparent 头格式：version-traceId-parentId-traceFlags
// 示例：00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01

class TraceContextPropagator {
  // 注入 Trace Context 到请求头
  inject(context: Context, carrier: any, setter: SetterFunction) {
    const spanContext = trace.getSpanContext(context);
    
    if (!spanContext || !isSpanContextValid(spanContext)) {
      return;
    }
    
    // 构建 traceparent 头
    const traceParent = `00-${spanContext.traceId}-${spanContext.spanId}-${spanContext.traceFlags}`;
    setter.set(carrier, 'traceparent', traceParent);
    
    // 构建 tracestate 头（可选，用于厂商特定信息）
    const traceState = spanContext.traceState?.serialize();
    if (traceState) {
      setter.set(carrier, 'tracestate', traceState);
    }
  }
  
  // 从请求头提取 Trace Context
  extract(context: Context, carrier: any, getter: GetterFunction): Context {
    const traceParent = getter.get(carrier, 'traceparent');
    
    if (!traceParent) {
      return context;
    }
    
    // 解析 traceparent
    const parts = traceParent.split('-');
    
    if (parts.length !== 4) {
      return context;
    }
    
    const [version, traceId, spanId, traceFlags] = parts;
    
    // 验证版本
    if (version !== '00') {
      return context;
    }
    
    // 验证格式
    if (!/^[a-f0-9]{32}$/.test(traceId)) {
      return context;
    }
    
    if (!/^[a-f0-9]{16}$/.test(spanId)) {
      return context;
    }
    
    // 创建 SpanContext
    const spanContext: SpanContext = {
      traceId,
      spanId,
      traceFlags: parseInt(traceFlags, 16),
      isRemote: true
    };
    
    // 创建 Span 并设置到 Context
    const span = new NonRecordingSpan(spanContext);
    return trace.setSpan(context, span);
  }
}

// HTTP 请求中的 Trace Context 传播示例
async function makeHttpRequest(url: string, context: Context) {
  const headers: Record = {};
  
  // 注入 Trace Context
  propagation.inject(context, headers, {
    set: (carrier, key, value) => {
      carrier[key] = value;
    }
  });
  
  // 发送 HTTP 请求
  const response = await fetch(url, { headers });
  
  return response;
}

// 服务端接收 Trace Context
async function handleRequest(req: Request) {
  // 提取 Trace Context
  const context = propagation.extract(
    ROOT_CONTEXT,
    req.headers,
    {
      get: (carrier, key) => carrier[key],
      keys: (carrier) => Object.keys(carrier)
    }
  );
  
  // 在提取的 Context 中创建 Span
  const span = tracer.startSpan('process-request', {}, context);
  
  try {
    // 处理请求
    await processRequest(req);
    span.setStatus({ code: SpanStatusCode.OK });
  } catch (error) {
    span.setStatus({ code: SpanStatusCode.ERROR, message: error.message });
    span.recordException(error);
    throw error;
  } finally {
    span.end();
  }
}
                        

import { trace, context, SpanKind, SpanStatusCode } from '@opentelemetry/api';

const tracer = trace.getTracer('agent-tracer', '1.0.0');

class AgentTracer {
  // 追踪 Agent 对话
  async traceConversation(
    userId: string,
    message: string
  ): Promise {
    return tracer.startActiveSpan(
      'agent-conversation',
      { kind: SpanKind.SERVER },
      async (span) => {
        // 设置 Tags
        span.setAttributes({
          'user.id': userId,
          'message.length': message.length,
          'conversation.type': 'customer-support'
        });
        
        try {
          // 1. 理解用户意图
          const intent = await this.traceIntentAnalysis(span, message);
          
          // 2. 检索知识库
          const knowledge = await this.traceKnowledgeRetrieval(span, intent);
          
          // 3. 调用 LLM 生成回复
          const response = await this.traceLLMCompletion(span, intent, knowledge);
          
          // 4. 后处理与验证
          const finalResponse = await this.tracePostProcessing(span, response);
          
          span.setStatus({ code: SpanStatusCode.OK });
          return finalResponse;
          
        } catch (error) {
          span.setStatus({ 
            code: SpanStatusCode.ERROR, 
            message: error.message 
          });
          span.recordException(error);
          throw error;
        } finally {
          span.end();
        }
      }
    );
  }
  
  // 追踪意图分析
  private async traceIntentAnalysis(parentSpan: any, message: string) {
    return tracer.startActiveSpan(
      'intent-analysis',
      { kind: SpanKind.INTERNAL },
      async (span) => {
        span.setAttributes({
          'analysis.model': 'intent-classifier-v2',
          'message.preview': message.substring(0, 100)
        });
        
        try {
          // 模拟意图分析
          await new Promise(resolve => setTimeout(resolve, 50));
          
          const intent = {
            category: 'billing',
            confidence: 0.92,
            entities: ['invoice', 'payment']
          };
          
          span.setAttributes({
            'intent.category': intent.category,
            'intent.confidence': intent.confidence
          });
          
          span.setStatus({ code: SpanStatusCode.OK });
          return intent;
          
        } catch (error) {
          span.setStatus({ code: SpanStatusCode.ERROR });
          throw error;
        } finally {
          span.end();
        }
      }
    );
  }
  
  // 追踪知识检索
  private async traceKnowledgeRetrieval(parentSpan: any, intent: any) {
    return tracer.startActiveSpan(
      'knowledge-retrieval',
      { kind: SpanKind.CLIENT },
      async (span) => {
        span.setAttributes({
          'retrieval.system': 'vector-database',
          'retrieval.topK': 5
        });
        
        try {
          // 模拟向量检索
          await new Promise(resolve => setTimeout(resolve, 120));
          
          const knowledge = [
            { id: 'doc-1', score: 0.89 },
            { id: 'doc-2', score: 0.85 }
          ];
          
          span.setAttributes({
            'retrieval.results.count': knowledge.length
          });
          
          span.setStatus({ code: SpanStatusCode.OK });
          return knowledge;
          
        } catch (error) {
          span.setStatus({ code: SpanStatusCode.ERROR });
          throw error;
        } finally {
          span.end();
        }
      }
    );
  }
  
  // 追踪 LLM 调用
  private async traceLLMCompletion(
    parentSpan: any, 
    intent: any, 
    knowledge: any[]
  ) {
    return tracer.startActiveSpan(
      'llm-completion',
      { kind: SpanKind.CLIENT },
      async (span) => {
        span.setAttributes({
          'llm.provider': 'openai',
          'llm.model': 'gpt-4-turbo',
          'llm.temperature': 0.7,
          'llm.max_tokens': 1024
        });
        
        try {
          // 模拟 LLM 调用
          await new Promise(resolve => setTimeout(resolve, 1500));
          
          const response = {
            content: 'Here is the answer...',
            usage: {
              prompt_tokens: 256,
              completion_tokens: 128,
              total_tokens: 384
            }
          };
          
          span.setAttributes({
            'llm.token_count': response.usage.total_tokens,
            'llm.response.length': response.content.length
          });
          
          // 记录事件
          span.addEvent('llm.response.received', {
            'response.tokens': response.usage.total_tokens
          });
          
          span.setStatus({ code: SpanStatusCode.OK });
          return response;
          
        } catch (error) {
          span.setStatus({ code: SpanStatusCode.ERROR });
          throw error;
        } finally {
          span.end();
        }
      }
    );
  }
  
  // 追踪后处理
  private async tracePostProcessing(parentSpan: any, response: any) {
    return tracer.startActiveSpan(
      'post-processing',
      { kind: SpanKind.INTERNAL },
      async (span) => {
        try {
          // 模拟后处理（内容过滤、格式化等）
          await new Promise(resolve => setTimeout(resolve, 30));
          
          span.setStatus({ code: SpanStatusCode.OK });
          return response.content;
          
        } catch (error) {
          span.setStatus({ code: SpanStatusCode.ERROR });
          throw error;
        } finally {
          span.end();
        }
      }
    );
  }
}

// 使用示例
const agentTracer = new AgentTracer();

// 处理用户请求
async function handleUserMessage(userId: string, message: string) {
  try {
    const response = await agentTracer.traceConversation(userId, message);
    return response;
  } catch (error) {
    console.error('Conversation failed:', error);
    throw error;
  }
}
                        

# fluent.conf 配置文件

# 全局配置

  log_level info
  workers 4


# 输入：采集容器日志

  @type tail
  path /var/log/containers/*.log
  pos_file /var/log/fluentd-containers.log.pos
  tag kubernetes.*
  read_from_head true
  
  
    @type json
    time_key time
    time_format %Y-%m-%dT%H:%M:%S.%NZ
  


# 过滤器：增强日志字段

  @type record_transformer
  
  
    hostname ${hostname}
    environment production
    cluster_id cluster-001
  


# 过滤器：解析 Agent 日志

  @type parser
  key_name log
  reserve_data true
  
  
    @type json
  


# 输出：发送到 Elasticsearch

  @type elasticsearch
  host elasticsearch.logging.svc
  port 9200
  logstash_format true
  logstash_prefix agent-logs
  logstash_dateformat %Y.%m.%d
  
  
    @type file
    path /var/log/fluentd/buffer
    flush_mode interval
    flush_interval 5s
    chunk_limit_size 8MB
    queue_limit_length 512
    retry_max_interval 30s
    retry_forever true
  


# 输出：错误日志发送到告警系统

  @type copy
  
  
    @type elasticsearch
    # ES 配置同上
  
  
  
    @type http
    endpoint https://alert-system.example.com/logs
    
      @type json
    
    
    
      @type memory
      flush_interval 3s
      retry_max_interval 10s
    
  

                        

// 查询最近 1 小时的 ERROR 日志
GET /agent-logs-*/_search
{
  "query": {
    "bool": {
      "must": [
        { "term": { "level": "ERROR" } },
        {
          "range": {
            "timestamp": {
              "gte": "now-1h",
              "lte": "now"
            }
          }
        }
      ]
    }
  },
  "sort": [
    { "timestamp": { "order": "desc" } }
  ],
  "size": 100
}

// 按用户 ID 聚合错误日志
GET /agent-logs-*/_search
{
  "size": 0,
  "query": {
    "term": { "level": "ERROR" }
  },
  "aggs": {
    "errors_by_user": {
      "terms": {
        "field": "user_id",
        "size": 10
      },
      "aggs": {
        "error_types": {
          "terms": {
            "field": "error_code"
          }
        }
      }
    }
  }
}

// 查询特定 Trace 的所有日志
GET /agent-logs-*/_search
{
  "query": {
    "term": { "trace_id": "4bf92f3577b34da6a3ce929d0e0e4736" }
  },
  "sort": [
    { "timestamp": { "order": "asc" } }
  ]
}

// 统计每 5 分钟的日志量
GET /agent-logs-*/_search
{
  "size": 0,
  "query": {
    "range": {
      "timestamp": { "gte": "now-24h" }
    }
  },
  "aggs": {
    "logs_over_time": {
      "date_histogram": {
        "field": "timestamp",
        "calendar_interval": "5m"
      },
      "aggs": {
        "log_levels": {
          "terms": {
            "field": "level"
          }
        }
      }
    }
  }
}