用FastAPI构建语义缓存代理，削减58%的大模型账单

生产环境中的大模型部署藏着一个不太体面的真相：在企业级SaaS应用里，30%到60%的LLM查询其实是重复或轻微改写的内容。客服对话、重复性的智能体任务、自动化数据丰富作业、文档问答流程，这些场景反复用功能上完全相同的提示词去冲击上游推理服务（OpenAI、Anthropic、Mistral）。

结果就是：高到离谱的月度推理账单、难以预测的延迟尖峰，以及工程团队在缺乏细粒度、按租户划分的可观测性时的两眼一抹黑。

本文将拆解如何用异步Python（FastAPI）、Redis向量检索做语义缓存、以及OpenTelemetry（OTel）统一追踪与token指标，搭建一个生产级、即插即用、兼容OpenAI接口的反向代理。

瓶颈：脆弱的商业网关代价高昂

当工程负责人意识到LLM开销正在失控时，通常会评估两条路径，但两者都有明显缺陷。

第一条是专有的可观测性与网关平台。商业APM和专门的LLM SaaS网关要么按席位收取经常性费用，要么在每百万token上加收溢价。更棘手的是，把未经脱敏的租户提示词发送给第三方网关，会带来严重的数据隐私与合规风险。

第二条是朴素的精确匹配缓存。直接对提示词做哈希（SHA256(prompt)）存入Redis，命中率低得可怜——往往不到4%。因为哪怕多出一个空格或标点，缓存就会失效。

Traditional Flow:
Client ──> Commercial Gateway ($$$ per token) ──> OpenAI API ($$$ per run)

Optimized Self-Hosted Flow:
Client ──> FastAPI Proxy ──> Redis Vector Search (Cosine >= 0.92) ──> Cache HIT (0ms upstream cost)
                    │ (Cache MISS)
                    └───> Upstream Provider (OTel Metric Exported)

可持续的解法是自建一个代理：它能作为精确的即插即用替代品（base_url="http://proxy-host/v1"），通过向量嵌入验证语义相似度，并把细粒度的OTel span直接送入Grafana、Jaeger或Prometheus。

五阶段架构流水线

整个代理管线按顺序执行五个阶段，每个阶段都承担明确职责。

拦截与租户鉴权：从请求头提取Authorization、X-Tenant-ID，并对照token预算配额做校验。

嵌入与相似度探测：用轻量级嵌入模型（例如text-embedding-3-small或本地部署的SentenceTransformer）把进入的用户提示词向量化。

Redis向量相似度查询：对已索引的提示词嵌入执行K近邻（KNN）搜索，用余弦相似度衡量。如果similarity >= threshold（例如0.92），立即返回缓存的补全结果。

熔断与token护栏：若缓存未命中，先核对租户实时的token消耗桶。如果滑动窗口预算已超额，直接返回HTTP 429，根本不触碰上游。

OTel span收尾：发出OTLP指标（提示token、补全token、延迟、缓存状态），并直接导入APM后端。

这套流程的核心价值在于——命中缓存时上游成本为零，而未命中时又有预算护栏兜底，避免账单失控。

核心代码与实现逻辑

Redis语义缓存索引初始化

首先要在Redis中建立一个支持HNSW向量检索的索引。这里采用1536维（对应text-embedding-3-small），距离度量选用余弦相似度。

import numpy as np
from redis.asyncio import Redis
from redis.commands.search.field import VectorField, TextField
from redis.commands.search.indexDefinition import IndexDefinition, IndexType
from redis.commands.search.query import Query

INDEX_NAME = "idx:semantic_cache"
VECTOR_DIM = 1536  # text-embedding-3-small dimension

async def init_redis_indices(redis_client: Redis):
    try:
        await redis_client.ft(INDEX_NAME).info()
    except Exception:
        schema = (
            TextField("prompt_text"),
            TextField("completion_text"),
            VectorField(
                "prompt_vector",
                "HNSW",
                {"TYPE": "FLOAT32", "DIM": VECTOR_DIM, "DISTANCE_METRIC": "COSINE"}
            )
        )
        definition = IndexDefinition(prefix=["cache:prompt:"], index_type=IndexType.HASH)
        await redis_client.ft(INDEX_NAME).create_index(schema, definition=definition)

HNSW（Hierarchical Navigable Small World）是目前工程实践中最主流的近似最近邻（ANN）索引算法之一。它的核心思想是构建一个多层图结构：顶层节点稀疏、跨度大，用于快速粗定位；底层节点密集、跨度小，用于精确搜索。查询时从顶层入口点出发，贪心地向更近邻居跳跃，逐层下沉，最终在底层找到候选集。与暴力KNN相比，HNSW在百万级向量上的查询延迟可低至个位数毫秒，召回率通常能保持在95%以上。代价是索引构建时的内存占用较高（每个向量节点需要存储多条边），以及索引构建时间比IVF类算法更长。Redis Stack内置的redis-search模块从7.2版本起原生支持HNSW，可以通过M（每层最大出边数，典型值16）和EF_CONSTRUCTION（构建时的动态候选集大小，典型值200）两个参数在召回率与内存间取得平衡。生产场景下建议将EF_RUNTIME（查询时动态候选集大小）设置为10到50之间，以在延迟与精度间找到合适的工作点。

语义匹配引擎

匹配引擎执行KNN搜索，取出最近的一条向量。需要特别注意余弦距离与相似度的换算：距离为0表示完全一致，为2表示完全相反，因此相似度 = 1 - 距离。

async def check_semantic_cache(redis_client: Redis, query_vector: list[float], threshold: float = 0.92):
    query_bytes = np.array(query_vector, dtype=np.float32).tobytes()
    query = (
        Query("*=>[KNN 1 @prompt_vector $vec AS score]")
        .sort_by("score")
        .return_fields("completion_text", "score")
        .dialect(2)
    )
    results = await redis_client.ft(INDEX_NAME).search(query, query_params={"vec": query_bytes})
    if results.docs:
        doc = results.docs[0]
        distance = float(doc.score)
        similarity = 1.0 - distance
        if similarity >= threshold:
            return doc.completion_text, similarity
    return None, 0.0

余弦相似度衡量的是两个向量之间的方向夹角，而非欧氏距离。计算公式为两向量点积除以各自模长之积，结果落在[-1, 1]区间，1表示方向完全一致，0表示正交，-1表示方向相反。对于文本嵌入而言，语义相近的句子其向量夹角通常很小，余弦相似度接近1。Redis返回的score字段采用的是余弦距离（1 - 余弦相似度），因此代码中需要做similarity = 1.0 - distance的换算，这是实现中容易出错的细节。选择0.92作为默认阈值意味着：只有当两段文本的语义重叠度极高时才触发缓存命中，这在实践中对应"问题的措辞不同但意图完全一致"的场景（如"怎么重置密码"与"如何找回密码"）。阈值过低会导致语义不相关的查询被误判命中（精度下降），阈值过高则接近精确匹配、缓存命中率大幅萎缩——因此阈值调优本质上是精度与召回率之间的工程权衡。

带OpenTelemetry埋点的代理路由

代理路由完整复刻了/v1/chat/completions接口。它先计算嵌入、查缓存，命中则直接返回（token计数记为0），未命中才转发到上游OpenAI，并把结果异步写回缓存。关键细节是：每一步都往span里写属性（租户ID、是否命中、余弦相似度），token计数器也按租户与命中状态分别打标签。

@app.post("/v1/chat/completions")
async def chat_completions_proxy(request: Request):
    tenant_id = request.headers.get("X-Tenant-ID", "default_tenant")
    payload = await request.json()
    messages = payload.get("messages", [])
    last_prompt = messages[-1]["content"] if messages else ""

    with tracer.start_as_current_span("llm_completion_router") as span:
        span.set_attribute("llm.tenant_id", tenant_id)
        prompt_vec = await compute_embedding(last_prompt)
        cached_response, similarity = await check_semantic_cache(app.state.redis, prompt_vec, threshold=0.92)
        if cached_response:
            span.set_attribute("llm.cache_hit", True)
            span.set_attribute("llm.cosine_similarity", similarity)
            token_counter.add(0, {"tenant": tenant_id, "cache_hit": "true"})
            return {"id": "cached-completion", "object": "chat.completion", "choices": [...], "usage": {"total_tokens": 0}}
        # Cache Miss - Forward to OpenAI
        ...

命中缓存意味着对上游的调用成本为零，这正是削减账单的核心机制。

OpenTelemetry（OTel）是CNCF主导的可观测性标准，统一了Traces（链路追踪）、Metrics（指标）和Logs（日志）三类信号的采集与导出协议，旨在替代厂商锁定的APM SDK。其核心概念包括：Span（一次操作的执行记录，含开始时间、结束时间、属性键值对和状态）、Trace（由因果关联的Span组成的有向无环图，代表一次完整请求链路）、以及OTLP（OpenTelemetry Line Protocol，标准化的数据导出协议，支持gRPC和HTTP两种传输方式）。在本代理架构中，每次/v1/chat/completions请求对应一个根Span，嵌入计算、缓存查询、上游转发各自作为子Span嵌套其中。通过在Span上打llm.tenant_id、llm.cache_hit、llm.cosine_similarity等属性，下游的Grafana或Jaeger可以按租户聚合分析缓存命中率和token消耗趋势。这种"代码即仪表盘数据源"的模式，使得可观测性从事后排障工具变为实时成本控制的决策依据。

部署与真实场景的边界处理

把语义代理推向生产，有三类现实中必须妥善处理的边界情况。

滑动窗口熔断器：如果某个自动化工作流失控（例如n8n的webhook陷入死循环），单纯的速率限制救不了你的账单。应该用原子性的Redis INCRBY操作，以租户加当前分钟为键（tenant:{id}:budget:{YYYYMMDDHHmm}）累计消耗。一旦超过token上限，立即熔断并返回HTTP 429。

动态相似度阈值：并非所有业务都能容忍宽松的语义匹配。客服FAQ机器人在0.88的阈值下工作良好，而法律或金融抽取任务则需要0.97甚至严格的确定性命中。因此阈值应该可通过HTTP头配置（X-Semantic-Threshold: 0.95）。

连接池与下游超时：给HTTPX客户端池设置严格的连接超时（通常2.0秒）与读取超时（30.0秒）。同时确保向量嵌入异步执行，让缓存查找的开销控制在总延迟12毫秒以内。

这三点看似琐碎，却决定了代理在真实流量压力下是资产还是负债。

结语

到这里，你已经掌握了一套完整的架构蓝图：消除冗余API调用、强制执行按租户的token护栏、导出企业级OpenTelemetry指标，而且不必为软件席位付费。

声称削减58%账单的核心前提，是企业场景里本就有相当比例的重复查询——语义缓存把这部分流量从昂贵的上游调用中拦截下来。阈值设定、熔断策略和异步嵌入的工程细节，共同决定了实际能省下多少。对于重度依赖LLM推理的团队，这类自建代理既能控成本，又能把数据主权和可观测性牢牢握在自己手里。

背景补充

Redis的INCRBY命令是原子操作，即便在高并发下多个代理实例同时写入同一键，计数结果也不会出现竞态条件导致的漏计或重复计数。结合EXPIRE命令（设置键的TTL为60或120秒），可以构造出一个轻量级的固定窗口计数器：以tenant:{id}:budget:{YYYYMMDDHHmm}为键，每次请求原子地累加token消耗，并在键首次创建时设置过期时间，窗口结束后键自动消失、下一分钟重新计数。"滑动窗口"则需要额外维护多个时间桶并汇总，实现更平滑的限流效果但逻辑稍复杂。两者都远比在应用层用内存计数更可靠，因为代理通常以多副本方式部署，本地内存计数器无法跨实例共享状态，会导致限额实际上被放大N倍（N为副本数）。