# main.py
from fastapi import FastAPI,
Request
from openai import OpenAI
import metrics
import asyncio
from contextlib import asynccontextmanager
import os
from typing import Dict,
Any
# 假设的模型成本表（美元 per 1K tokens）
MODEL_COST = {
"gpt-3.5-turbo":
{"prompt": 0.0015, "completion":
0.002},
"gpt-4": {"prompt":
0.03, "completion": 0.06}
}
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
app = FastAPI()
# 简单的追踪上下文模拟（生产环境应使用OpenTelemetry）
class TraceContext:
def __init__(self, trace_id:
str):
self.trace_id = trace_id
self.spans
= []
def start_span(self, name:
str):
span = {"name":
name, "start": time.time()}
self.spans.append(span)
return span
def end_span(self, span:
Dict[str, Any]):
span["end"] = time.time()
span["duration"]
= span["end"] - span["start"]
@app.middleware("http")
async def add_metrics_and_trace(request:
Request, call_next):
"""全局中间件：记录请求耗时和初始化追踪"""
start_time = time.time()
# 为每个请求生成一个追踪ID
trace_id = f"trace_{int(start_time*1000)}_{hash(request.url.path)}"
request.state.trace = TraceContext(trace_id)
response = await call_next(request)
duration = time.time() -
start_time
metrics.APP_REQUEST_DURATION.observe(duration)
# 可以在此处将Trace信息发送到后端（如Jaeger）
# print(f"Trace {trace_id}
completed with {len(request.state.trace.spans)}
spans.")
return response
@app.post("/chat")
async def chat_completion(request:
Request, user_input: str):
"""处理聊天请求，集成指标收集和简单追踪"""
trace = request.state.trace
# Span 1: 请求预处理
span_preprocess = trace.start_span("preprocess")
# ... 可能的输入清洗、意图识别
await asyncio.sleep(0.01)
# 模拟处理
trace.end_span(span_preprocess)
# Span 2: LLM调用 (核心监控点)
span_llm = trace.start_span("llm_call")
model = "gpt-3.5-turbo"
# 记录请求开始
metrics.LLM_REQUESTS_TOTAL.labels(model=model,
endpoint="chat.completions").inc()
llm_start = time.time()
try:
response = client.chat.completions.create(
model=model,
messages=[{"role":
"user", "content": user_input}],
temperature=0.7,
)
status = "success"
except Exception as e:
status = "error"
raise e
finally:
# 记录延迟和状态
llm_duration = time.time()
- llm_start
metrics.LLM_REQUEST_DURATION.labels(model=model,
status=status).observe(llm_duration)
trace.end_span(span_llm)
if status == "success":
completion = response.choices[0].message.content
usage = response.usage
# 记录Token用量和成本
metrics.LLM_TOKENS_USED.labels(model=model,
token_type="prompt").inc(usage.prompt_tokens)
metrics.LLM_TOKENS_USED.labels(model=model,
token_type="completion").inc(usage.completion_tokens)
estimated_cost = (usage.prompt_tokens
/ 1000 * MODEL_COST[model]["prompt"]
+
usage.completion_tokens /
1000 * MODEL_COST[model]["completion"])
metrics.LLM_REQUEST_COST.labels(model=model).inc(estimated_cost)
# Span 3: 后处理
span_post = trace.start_span("postprocess")
# ... 可能的输出格式化、敏感信息过滤
await asyncio.sleep(0.005)
trace.end_span(span_post)
return {
"response": completion,
"usage": usage.dict(),
"estimated_cost_usd":
round(estimated_cost, 6),
"trace_id": trace.trace_id
}
# 添加Prometheus指标端点
app.add_route("/metrics",
metrics.metrics_endpoint)
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0",
port=8000)
订阅