C16 · AI 服务生产栈 Python 侧

// Java / Spring Boot 3：Python 服务只暴露 HTTP 契约，Java 侧不感知其运行时
@RestController
public class InferenceController {
  private final RestClient python;

  public InferenceController(RestClient.Builder b,
                             @Value("${inference.python.base-url}") String url) {
    this.python = b.baseUrl(url)
        .requestFactory(new SimpleClientHttpRequestFactory() {{
          setConnectTimeout(Duration.ofMillis(300));
          setReadTimeout(Duration.ofSeconds(8));
        }})
        .build();
  }

  @PostMapping("/v1/ask")
  public Answer ask(@RequestBody AskRequest req) {
    return python.post().uri("/v1/generate")
        .body(new GenerateRequest(req.question(), req.traceId()))
        .retrieve()
        .onStatus(s -> s.is5xxServerError(), (rq, rs) -> {
          throw new InferenceUnavailableException(rs.getStatusCode());
        })
        .body(Answer.class);
  }
}
public record GenerateRequest(String question, String traceId) {}
public record Answer(String text, int promptTokens, long latencyMs) {}

# Python：最小服务入口（依赖见 C16.02）
from fastapi import FastAPI
from pydantic import BaseModel, Field

app = FastAPI(title="inference-python")

class GenerateRequest(BaseModel):
    question: str = Field(min_length=1, max_length=4096)
    trace_id: str | None = None

class Answer(BaseModel):
    text: str
    prompt_tokens: int
    latency_ms: int

@app.get("/healthz")
def healthz() -> dict[str, str]:
    return {"status": "ok", "runtime": "cpython"}

@app.post("/v1/generate", response_model=Answer)
def generate(req: GenerateRequest) -> Answer:
    # 真实实现委托给推理引擎；此处保留契约与计时边界
    return Answer(text="", prompt_tokens=0, latency_ms=0)

fastapi==0.115.6
uvicorn[standard]==0.34.0
pydantic==2.10.4
vllm==0.6.6.post1
prometheus-client==0.21.1
opentelemetry-sdk==1.29.0
opentelemetry-exporter-otlp==1.29.0
httpx==0.28.1
pytest==8.3.4

# app.py
import time, os
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from prometheus_client import Counter, Histogram, generate_latest, CONTENT_TYPE_LATEST
from fastapi.responses import Response

app = FastAPI()
REQS = Counter("infer_requests_total", "requests", ["model", "status"])
LAT = Histogram("infer_latency_seconds", "latency", buckets=(.1, .5, 1, 2, 5))

class Req(BaseModel):
    prompt: str = Field(min_length=1, max_length=8192)
    max_tokens: int = Field(default=256, ge=1, le=4096)

class Resp(BaseModel):
    text: str
    prompt_tokens: int
    latency_ms: int

# vLLM 引擎在进程启动时加载一次，避免每请求重复加载权重
from vllm import LLM, SamplingParams
engine = LLM(model=os.environ["MODEL_PATH"],
             gpu_memory_utilization=0.90,
             max_model_len=8192,
             max_num_seqs=64)

@app.get("/healthz")
def healthz():
    return {"status": "ok", "model": os.environ["MODEL_PATH"]}

@app.get("/metrics")
def metrics():
    return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)

@app.post("/v1/generate", response_model=Resp)
def generate(req: Req):
    t0 = time.perf_counter()
    try:
        out = engine.generate([req.prompt],
                              SamplingParams(max_tokens=req.max_tokens, temperature=0.2))
        REQS.labels("vllm", "ok").inc()
        return Resp(text=out[0].outputs[0].text,
                    prompt_tokens=len(out[0].prompt_token_ids),
                    latency_ms=int((time.perf_counter() - t0) * 1000))
    except Exception:
        REQS.labels("vllm", "error").inc()
        raise HTTPException(status_code=503, detail="inference failed")
    finally:
        LAT.observe(time.perf_counter() - t0)

# 启动：uvicorn app:app --host 0.0.0.0 --port 8000 --workers 1
# vLLM 自带连续批处理，务必单 worker；多 worker 会重复占用显存

# Dockerfile
FROM nvidia/cuda:12.4.1-runtime-ubuntu22.04
WORKDIR /srv
RUN apt-get update && apt-get install -y --no-install-recommends python3.11 python3-pip \
    && rm -rf /var/lib/apt/lists/*
COPY requirements.txt .
RUN pip3 install --no-cache-dir -r requirements.txt
COPY app.py .
ENV MODEL_PATH=/models/qwen2.5-7b-instruct
EXPOSE 8000
HEALTHCHECK --interval=15s --timeout=3s --retries=3 \
  CMD python3 -c "import urllib.request;urllib.request.urlopen('http://127.0.0.1:8000/healthz')"
CMD ["python3", "-m", "uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]

# test_app.py —— CI 无 GPU，导入前用替身替换推理引擎
import sys, types
from fastapi.testclient import TestClient

class FakeOut:
    prompt_token_ids = [1, 2, 3]
    outputs = [type("O", (), {"text": "ok"})()]

fake = types.ModuleType("vllm")
fake.LLM = lambda *a, **k: types.SimpleNamespace(generate=lambda *a, **k: [FakeOut()])
fake.SamplingParams = lambda **k: k
sys.modules["vllm"] = fake

from app import app          # 此时 LLM(...) 不会真加载权重
client = TestClient(app)

def test_healthz():
    assert client.get("/healthz").json()["status"] == "ok"

def test_generate_contract():
    r = client.post("/v1/generate", json={"prompt": "hi", "max_tokens": 8})
    assert r.status_code == 200
    body = r.json()
    assert {"text", "prompt_tokens", "latency_ms"} <= body.keys()
    assert body["text"] == "ok"

# 运行：pip install -r requirements.txt && pytest -q test_app.py

# 显存粗算：权重 + KV Cache + 激活 + 框架预留
def estimate_weights_gb(params_b: float, dtype_bytes: int = 2) -> float:
    return params_b * dtype_bytes  # 7B FP16 约 14 GB

def kv_cache_gb(layers: int, heads: int, head_dim: int, seq_len: int,
                batch: int, dtype_bytes: int = 2) -> float:
    # 2 = K + V
    return 2 * layers * heads * head_dim * seq_len * batch * dtype_bytes / 1024**3

if __name__ == "__main__":
    w = estimate_weights_gb(7.0)
    kv = kv_cache_gb(layers=28, heads=28, head_dim=128, seq_len=8192, batch=64)
    print(f"weights={w:.1f}GB kv={kv:.1f}GB 需显存 >= {w + kv + 2:.1f}GB")

# 加载闸门：AWQ/GPTQ 量化把权重压到 4bit，并限制 KV 占用
from vllm import LLM
llm = LLM(model="/models/qwen2.5-7b-instruct-awq",
          quantization="awq",
          gpu_memory_utilization=0.85,   # 预留给激活与碎片
          max_model_len=4096,            # 直接约束 KV 上限
          enforce_eager=False,
          swap_space=4)

# vLLM 动态批：三个参数决定「吞吐 / 时延 / 显存」三角
from vllm import LLM, SamplingParams
llm = LLM(model="/models/qwen2.5-7b-instruct",
          max_num_seqs=128,              # 并发序列上限
          max_num_batched_tokens=16384,  # 单步可打包的 token 上限
          enable_chunked_prefill=True,   # 长 prompt 分块，避免阻塞 decode
          gpu_memory_utilization=0.90)

import time
prompts = [f"请总结第 {i} 段文档" for i in range(200)]
t0 = time.perf_counter()
outs = llm.generate(prompts, SamplingParams(max_tokens=128, temperature=0))
elapsed_seconds = time.perf_counter() - t0
total_tokens = sum(len(o.outputs[0].token_ids) for o in outs)
print("output tok/s ≈", total_tokens / elapsed_seconds)

# 压测对照：固定并发梯度，观察 p99 与吞吐拐点
# python3 -m vllm.entrypoints.openai.api_server ... &
# hey -z 60s -c 32 -m POST -H 'Content-Type: application/json' \
#     -d '{"model":"qwen2.5-7b-instruct","prompt":"hi","max_tokens":64}' \
#     http://127.0.0.1:8000/v1/completions

# pip install pymilvus sentence-transformers langchain-text-splitters
from langchain_text_splitters import RecursiveCharacterTextSplitter
from sentence_transformers import SentenceTransformer
from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType

splitter = RecursiveCharacterTextSplitter(chunk_size=512, chunk_overlap=64,
                                          separators=["\n\n", "\n", "。", ""])
encoder = SentenceTransformer("BAAI/bge-m3")
connections.connect(alias="default", host="milvus", port="19530")

fields = [
    FieldSchema(name="id", dtype=DataType.VARCHAR, max_length=64, is_primary=True),
    FieldSchema(name="doc_id", dtype=DataType.VARCHAR, max_length=64),
    FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=2048),
    FieldSchema(name="vec", dtype=DataType.FLOAT_VECTOR, dim=1024),
]
schema = CollectionSchema(fields, description="kb_chunks")
col = Collection("kb_chunks", schema=schema)
col.create_index("vec", {"index_type": "HNSW", "metric_type": "COSINE",
                         "params": {"M": 16, "efConstruction": 200}})
col.load()

def ingest(doc_id: str, raw: str) -> int:
    chunks = splitter.split_text(raw)
    # 批量编码，避免逐条调用导致 GPU 空转
    vectors = encoder.encode(chunks, batch_size=64, normalize_embeddings=True)
    rows = [
        [f"{doc_id}-{i}" for i in range(len(chunks))],
        [doc_id] * len(chunks),
        chunks,
        vectors.tolist(),
    ]
    col.insert(rows)
    col.flush()
    return len(chunks)

# 检索：与 Java 侧 VectorStore 使用同一 metric / dim，否则相似度不可比
def search(q: str, top_k: int = 5):
    v = encoder.encode([q], normalize_embeddings=True).tolist()
    return col.search(v, "vec", {"metric_type": "COSINE", "params": {"ef": 64}},
                      limit=top_k, output_fields=["doc_id", "text"])

import asyncio, time
from concurrent.futures import ProcessPoolExecutor

async def call_llm(prompt: str) -> str:
    # 正确：异步 HTTP 客户端，等待期间事件循环可服务其他请求
    await asyncio.sleep(0.05)
    return prompt.upper()

async def cpu_bound(text: str) -> int:
    # 正确：CPU 密集交给进程池，绕开 GIL
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(POOL, len, text * 1000)

POOL = ProcessPoolExecutor(max_workers=4)

async def main():
    t0 = time.perf_counter()
    results = await asyncio.gather(*(call_llm(f"q{i}") for i in range(50)))
    print("50 并发耗时", round(time.perf_counter() - t0, 3), "s", len(results))
    # 反例：在协程里直接 time.sleep / 同步 requests，会串行化全部并发
    # for i in range(50): time.sleep(0.05)   # 耗时约 2.5s，且阻塞整个 loop

asyncio.run(main())

# GIL 陷阱复现：CPU 密集用线程反而更慢
import threading, time

def burn(n):
    s = 0
    for i in range(n):
        s += i * i
    return s

N = 20_000_000
t0 = time.perf_counter(); burn(N); burn(N)
print("串行", round(time.perf_counter() - t0, 2), "s")
t0 = time.perf_counter()
ts = [threading.Thread(target=burn, args=(N,)) for _ in range(2)]
[t.start() for t in ts]; [t.join() for t in ts]
print("双线程", round(time.perf_counter() - t0, 2), "s  # 不快反慢")

# 部署：CPU 密集用多进程；vLLM 已占用 GPU，禁止多 worker 重复加载权重
# gunicorn app:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000
# 若单卡 vLLM：uvicorn app:app --workers 1（靠引擎内部连续批处理提吞吐）

// proto/inference.proto
syntax = "proto3";
package ai.inference.v1;
option java_package = "com.example.ai.inference.grpc";
option java_outer_classname = "InferenceProto";

service InferenceService {
  rpc Generate (GenerateRequest) returns (GenerateReply);
}
message GenerateRequest {
  string trace_id = 1;      // 跨栈 Trace 透传
  string prompt = 2;
  int32 max_tokens = 3;
}
message GenerateReply {
  string text = 1;
  int32 prompt_tokens = 2;
  int64 latency_ms = 3;     // int64 在 JSON 中须转 string，见 C18.09
}

# python -m grpc_tools.protoc -I proto --python_out=. --grpc_python_out=. proto/inference.proto
from concurrent import futures
import grpc, time
import inference_pb2 as pb, inference_pb2_grpc as pbg

class Inference(pbg.InferenceServiceServicer):
    def Generate(self, request, context):
        t0 = time.perf_counter()
        if not request.prompt:
            context.abort(grpc.StatusCode.INVALID_ARGUMENT, "prompt required")
        # context 可透传 deadline；超时后引擎侧应停止生成，避免 GPU 空转
        text = f"echo:{request.prompt}"
        return pb.GenerateReply(
            text=text,
            prompt_tokens=len(request.prompt),
            latency_ms=int((time.perf_counter() - t0) * 1000),
        )

server = grpc.server(futures.ThreadPoolExecutor(max_workers=8),
                     options=[("grpc.max_receive_message_length", 8 * 1024 * 1024)])
pbg.add_InferenceServiceServicer_to_server(Inference(), server)
server.add_insecure_port("[::]:50051")
server.start(); server.wait_for_termination()

// Spring Boot 侧：由 proto 生成的阻塞 stub + Deadline
@Configuration
public class GrpcConfig {
  @Bean
  InferenceServiceGrpc.InferenceServiceBlockingStub stub(
      @Value("${inference.grpc.target}") String target) {
    ManagedChannel ch = ManagedChannelBuilder.forTarget(target)
        .usePlaintext()
        .enableRetry()
        .maxRetryAttempts(2)
        .build();
    return InferenceServiceGrpc.newBlockingStub(ch);
  }
}

@Service
public class InferenceGateway {
  private final InferenceServiceGrpc.InferenceServiceBlockingStub stub;
  InferenceGateway(InferenceServiceGrpc.InferenceServiceBlockingStub s) { this.stub = s; }

  public GenerateReply generate(String prompt, String traceId) {
    return stub.withDeadlineAfter(3, TimeUnit.SECONDS)
        .withInterceptors(MetadataUtils.newAttachHeadersInterceptor(meta(traceId)))
        .generate(GenerateRequest.newBuilder()
            .setTraceId(traceId).setPrompt(prompt).setMaxTokens(256).build());
  }
  private Metadata meta(String traceId) {
    Metadata m = new Metadata();
    m.put(Metadata.Key.of("x-trace-id", Metadata.ASCII_STRING_MARSHALLER), traceId);
    return m;
  }
}

# pip install opentelemetry-sdk opentelemetry-exporter-otlp langfuse
import os, time
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from prometheus_client import Histogram, Counter
from langfuse import Langfuse

provider = TracerProvider()
provider.add_span_processor(BatchSpanProcessor(
    OTLPSpanExporter(endpoint=os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"], insecure=True)))
trace.set_tracer_provider(provider)
tracer = trace.get_tracer("inference.python")

# LLMOps 平台：Run/Node 级 Trace、 Token 成本、评分统一回写
lf = Langfuse(public_key=os.environ["LANGFUSE_PK"],
              secret_key=os.environ["LANGFUSE_SK"],
              host=os.environ.get("LANGFUSE_HOST", "https://cloud.langfuse.com"))

LAT = Histogram("model_call_seconds", "model latency", ["model"])
TOK = Counter("model_tokens_total", "tokens", ["model", "kind"])

def generate(prompt: str, model: str) -> str:
    with tracer.start_as_current_span("llm.generate") as span:
        span.set_attribute("llm.model", model)
        span.set_attribute("llm.prompt_tokens", len(prompt))
        trace_obj = lf.trace(name="generate", input={"prompt": prompt})
        gen = trace_obj.generation(name="chat", model=model,
                                   model_parameters={"temperature": 0.2})
        t0 = time.perf_counter()
        try:
            text = do_call(prompt)             # 真实模型调用
            gen.end(output=text, usage={"input": len(prompt), "output": len(text)})
            trace_obj.score(name="quality", value=0.9)
            TOK.labels(model, "output").inc(len(text))
            return text
        finally:
            LAT.labels(model).observe(time.perf_counter() - t0)
            lf.flush()

import threading, time
from collections import defaultdict

RATE_PER_1K = {"qwen2.5-7b": 0.002, "qwen-max": 0.012}  # 单位：元 / 1K token

class BudgetGuard:
    def __init__(self, qps: int = 20, daily_quota: float = 500.0):
        self.qps, self.daily_quota = qps, daily_quota
        self.cost = defaultdict(float)
        self._sem = threading.BoundedSemaphore(qps)
        self._lock = threading.Lock()

    def acquire(self) -> bool:
        return self._sem.acquire(timeout=0.5)

    def charge(self, tenant: str, model: str, in_tok: int, out_tok: int) -> float:
        amount = (in_tok + out_tok) / 1000 * RATE_PER_1K[model]
        with self._lock:
            if self.cost[tenant] + amount > self.daily_quota:
                raise PermissionError("daily quota exceeded")
            self.cost[tenant] += amount
        return amount

    def release(self):
        self._sem.release()

# 使用：超限快速失败，避免 GPU 空转后才发现超预算
guard = BudgetGuard()
def handle(tenant, model, prompt):
    if not guard.acquire():
        raise RuntimeError("429 too many requests")
    try:
        text = do_call(prompt)
        cost = guard.charge(tenant, model, len(prompt), len(text))
        return {"text": text, "cost_cny": round(cost, 6)}
    finally:
        guard.release()