RAG在生产环境中的实际运行成本：完整成本分解

以下假设已完整陈述，以便模型可复现。语料库：10万份文档，平均每份1,500个token（1.5亿个token的文本），以512个token为一块、64个token重叠进行分块，产生约335,000个分块（1.71亿个嵌入token；重叠导致1.14倍的扩展）。每次查询：一个32个token的问题嵌入，对20个检索候选进行重排序（约10,400个输入token，40个输出），并对前5个分块进行答案生成（约2,900个输入token，350个输出）。模型：Qwen3 Embedding 0.6B（每百万token $0.04），用于重排序的DeepSeek V4 Flash（每百万 $0.068进 / $0.168出），用于答案生成的gpt-oss-120b（$0.10 / $0.70），备选Llama 3.3 70B（$0.65 / $0.65）。月份按30天计。

一次性摄取成本

摄取10万份文档，不做增强成本低于10美元，做增强则低于20美元。如果语料库翻倍，这些数字也会翻倍；它们永远不会成为问题。

三种流量水平下的月度成本

使用gpt-oss-120b的全包每次查询成本，是每个总计除以该层级的月查询量得出的：低流量时约$0.0033（$98 ÷ 30,000次查询，其中固定数据库成本占主导），中流量时约$0.0017（$496 ÷ 300,000），高流量时约$0.0013（$4,000 ÷ 3,000,000）。数据库套餐价格是当前具有代表性的档位；请根据你所在的区域和引擎，对照托管数据库套餐进行确认。

从表格中可以得出三点，并且每一点都与一个常见假设相矛盾。

这个模型中的token价格并不是异常低。对于gpt-oss-120b，Fireworks AI、Groq和Together AI在2026年7月都公布了每100万输入token 0.15美元、每100万输出token 0.60美元的价格，而DigitalOcean是0.10美元和0.70美元（经过验证的定价和兼容性差异见我们的OpenAI兼容推理API对比）。哪个更便宜取决于你的流量形态，而RAG有一种独特的形态：这条流水线每生成一个输出token大约发送8个输入token，因为模型要读取五个分块来写一个答案。输入密集的流量有利于输入费率更低的提供商。按照这个模型的token组合，在DigitalOcean上每次查询的生成成本为0.000535美元，而在0.15美元/0.60美元的提供商那里为0.000645美元：每天10万次查询时，每月为1,605美元对1,935美元。通过OpenRouter这样的聚合器路由，还会在服务请求的提供商费用之上额外增加5.5%的信用手续费。

更大的差异是结构性的，而不是按token计算的。RAG流水线不只是推理：它需要一个向量数据库，而大多数推理提供商并不运行向量数据库。在仅提供推理的提供商上构建这套技术栈，意味着需要第二个供应商来提供向量存储（专用向量数据库，或在租用计算资源上自托管pgvector），第二张账单，独立的访问控制，以及检索与生成之间的一个网络跳点，它位于每次查询的延迟路径上。在同一个账户中运行托管PostgreSQL和推理端点，消除了跨供应商跳点，将整个流水线的支出保留在一张发票上，并且意味着支撑你向量的数据库与其余基础设施拥有相同的备份、故障转移和监控。上面按token计算的差异是真实的，但很小；运营整合通常是更有力的理由。

以上所有内容都基于一条具体的流水线，本文其余部分将构建它。摄取部分运行一次（并在文档变化时再次运行）：读取文档，将其拆分为分块，为每个分块生成嵌入向量，并将分块文本和向量写入PostgreSQL。服务部分在每一个问题上运行：对问题进行嵌入，拉取最近的分块，对它们进行重排序，然后将幸存者传递给一个生成答案的模型。

Documents ──> Chunking ──> Embeddings API ──> Managed PostgreSQL (pgvector)
                                                        │
User question ──> Embeddings API ──> vector search ─────┘
                                          │
                                    top 20 chunks
                                          │
                                    LLM reranker
                                          │
                                     top 5 chunks
                                          │
                            Serverless Inference (answer)

所有与模型相关的操作都通过一个端点进行，https://inference.do-ai.run/v1，该端点与OpenAI兼容：修改基础URL和API密钥后，官方OpenAI SDK即可正常工作，切换模型只需更改一个字符串。整个流水线全程使用开放权重模型，这正是成本模型能保持上述水平的原因。

要继续操作，你需要一个DigitalOcean账户，并配备模型访问密钥和预付的无服务器推理余额、一个托管PostgreSQL集群，以及安装了openai、psycopg2-binary和tiktoken的Python 3.10+环境。

分块

语言模型和嵌入模型在聚焦的段落上比在完整文档上表现更好，向量搜索也以你嵌入的粒度进行检索。分块就是设置这种粒度的环节，它直接决定了成本模型中1.71亿token的嵌入量。

我们使用512个token的固定大小分块，并设置64个token的重叠。固定大小分块并非最聪明的策略，但它可预测、成本低且不易出错，这正是第一个生产版本所需要的。重叠的存在是为了让落在分块边界上的句子至少在某个分块中保持完整；这也是为什么1.5亿token的文本变成了1.71亿token的嵌入输入。重叠并不是免费的，这正是你需要为之付出代价的地方。

import tiktoken

enc = tiktoken.get_encoding("cl100k_base")

CHUNK_TOKENS = 512
OVERLAP_TOKENS = 64

def chunk_text(text: str, doc_id: str) -> list[dict]:
    tokens = enc.encode(text)
    stride = CHUNK_TOKENS - OVERLAP_TOKENS
    chunks = []
    for i, start in enumerate(range(0, len(tokens), stride)):
        window = tokens[start : start + CHUNK_TOKENS]
        if len(window) < 32:   # skip trailing fragments
            break
        chunks.append({
            "doc_id": doc_id,
            "chunk_index": i,
            "text": enc.decode(window),
        })
    return chunks

嵌入

嵌入将一段文本转换为向量，使得语义相似的文本在向量空间中彼此接近。正是这一点，使得“how do I reset my password”能够找到标题为“credential recovery procedure”的文本块，即使它们没有共同的关键词。

无服务器嵌入API提供了多个开源模型。我们使用Qwen3 Embedding 0.6B，价格为每百万输入词元$0.04：它能轻松处理长输入，在多语言和检索基准上表现良好，并生成1024维向量。如果你的语料库仅包含英文，且成本比质量更重要，BGE-M3和E5 Large V2可以以每百万词元$0.02的价格获得，这样会把本就低廉的$6.86费用再减半。

import os
from openai import OpenAI

client = OpenAI(
    base_url="https://inference.do-ai.run/v1",
    api_key=os.environ["DIGITALOCEAN_INFERENCE_KEY"],
)

def embed_batch(texts: list[str]) -> list[list[float]]:
    resp = client.embeddings.create(
        model="qwen3-embedding-0.6b",
        input=texts,
    )
    return [item.embedding for item in resp.data]

将数据分批发送（每批32到64条，可在吞吐量与请求大小之间取得平衡），并加入带指数退避的重试逻辑；任何跨越数十万次请求的摄取任务最终都会遇到瞬时故障。

可选：以半价批量推理增强数据块

原始数据块会丢失文档级别的上下文。例如，一个数据块写着“上限已提高到50”，但由于其中没有说明是什么的上限、在哪里，检索效果会很差。一个常见的解决办法是在嵌入每个数据块之前，先在其前面加上一行由LLM生成的简短上下文说明：说明它来自哪个文档以及内容是关于什么的。

这是一项离线、对延迟不敏感的工作，因此非常适合使用批量推理；关于折扣如何运作以及如何提交、监控和下载任务，在上述文章和批量推理指南中都有说明，因此这里我们只关注RAG特有的问题：这条流水线中哪些部分应该放入批处理任务。

规则很简单：将用户在到来之前执行的任何操作都进行批处理。数据块增强恰好符合这一条件，因为没有人等待它完成；任务在一小时还是二十四小时内完成，对流水线没有任何影响。查询时的工作（检索、重排序、答案生成）永远不符合条件，因为另一端有用户等待。有两个限制因素影响这种适配性：批量推理目前仅支持聊天端点上的OpenAI和Anthropic文本模型，因此即使流水线的其他部分使用开放权重模型，这一步骤也必须使用商业模型；此外，嵌入请求本身不能通过批量推理处理（按每百万token 0.04美元的价格，它们也没有必要这样做；LLM增强步骤才是摄取成本的主要去向，而非嵌入）。

批处理文件中的每个请求都会要求一个小模型来为一个数据块提供上下文定位：

def enrichment_request(chunk: dict, doc_title: str) -> dict:
    return {
        "custom_id": f'{chunk["doc_id"]}-{chunk["chunk_index"]}',
        "method": "POST",
        "url": "/v1/chat/completions",
        "body": {
            "model": "gpt-5-nano",
            "messages": [{
                "role": "user",
                "content": (
                    f"Document title: {doc_title}\n\n"
                    f"Chunk:\n{chunk['text']}\n\n"
                    "Write one sentence situating this chunk within the "
                    "document, for use as a search context prefix. "
                    "Output only the sentence."
                ),
            }],
            "max_tokens": 80,
        },
    }

对于335,000个数据块（使用GPT-5 nano时，每个请求大约包含650个输入token和60个输出token），该作业在无服务器模式下成本约为18.92美元，通过批量处理则为9.46美元：50%的折扣在唯一涉及大语言模型的摄取步骤中发挥了作用。在嵌入之前将返回的句子添加到每个数据块前，嵌入和重排序器都能看到新增的上下文。

此步骤是可选的。首先在不包含该步骤的情况下交付，查看检索失败的情况，如果数据块因缺乏上下文而检索效果不佳，再将其添加。之后重新嵌入将额外花费7美元，这足够便宜，可以将其视为一次实验而非长期承诺。

托管PostgreSQL中的向量存储

你需要一个地方来存储335,000个向量及其文本，并快速找到与查询最接近的向量。专用向量数据库是一种选择，但如果你已经在使用PostgreSQL，pgvector扩展可以将你现有的数据库转变为所需的向量存储，而备份、故障转移和访问控制已由托管平台处理。DigitalOcean托管PostgreSQL支持vector（pgvector）和vectorscale作为标准扩展。

CREATE EXTENSION IF NOT EXISTS vector;

CREATE TABLE chunks (
    id          BIGSERIAL PRIMARY KEY,
    doc_id      TEXT NOT NULL,
    chunk_index INT  NOT NULL,
    text        TEXT NOT NULL,
    embedding   vector(1024) NOT NULL,
    UNIQUE (doc_id, chunk_index)
);

CREATE INDEX chunks_embedding_idx
    ON chunks USING hnsw (embedding vector_cosine_ops);

HNSW索引是使搜索快速的原因。没有它，每个查询扫描所有335,000个向量；有了它，查询遍历图并在这种规模下以个位数毫秒返回。在批量加载数据之后而不是之前构建索引：插入到现有HNSW索引比在完整表上构建一次要慢得多。

摄取循环将前三个阶段连接在一起：

import psycopg2
from psycopg2.extras import execute_values

conn = psycopg2.connect(os.environ["DATABASE_URL"])

def store_chunks(chunks: list[dict], embeddings: list[list[float]]):
    rows = [
        (c["doc_id"], c["chunk_index"], c["text"], emb)
        for c, emb in zip(chunks, embeddings)
    ]
    with conn.cursor() as cur:
        execute_values(
            cur,
            """INSERT INTO chunks (doc_id, chunk_index, text, embedding)
               VALUES %s
               ON CONFLICT (doc_id, chunk_index)
               DO UPDATE SET text = EXCLUDED.text,
                             embedding = EXCLUDED.embedding""",
            rows,
            template="(%s, %s, %s, %s::vector)",
        )
    conn.commit()

对(doc_id, chunk_index)进行upsert操作可使重新摄取具有幂等性：重新运行一个文档，其块会被替换而非重复。当源中的文档被删除时，按doc_id删除其行。

在规模方面，这就是成本表中数据库行的来源：335,000个向量，1,024维，每维4字节，约1.4 GB，加上HNSW图和块文本，该表在磁盘上约4 GB。单节点托管PostgreSQL起价为每月$15，适合开发。对于生产环境，HNSW搜索性能取决于索引常驻内存，因此请选择RAM明显高于索引大小的计划，并添加备用节点以实现高可用性（每节点每月$30起）。额外存储为每月每GiB $0.21，对于此语料库大约合计$1。

检索

处理一个问题时，首先使用与摄取时相同的模型对其进行嵌入（这不是可选的；不同模型的向量位于不同的空间中，比较它们毫无意义），然后向PostgreSQL请求最近的块：

def retrieve(question: str, k: int = 20) -> list[dict]:
    q_emb = embed_batch([question])[0]
    with conn.cursor() as cur:
        cur.execute(
            """SELECT doc_id, chunk_index, text,
                      1 - (embedding <=> %s::vector) AS score
               FROM chunks
               ORDER BY embedding <=> %s::vector
               LIMIT %s""",
            (q_emb, q_emb, k),
        )
        cols = ["doc_id", "chunk_index", "text", "score"]
        return [dict(zip(cols, row)) for row in cur.fetchall()]

<=> 运算符是余弦距离，与 vector_cosine_ops 索引匹配。我们刻意过度获取：20 个文本块多于答案模型实际看到的数量，因为下一阶段的存在就是为了将真正相关的与仅相似的区分开来。

有两个值得了解的升级，但都不是第一版所必需的。混合搜索将向量相似性与 PostgreSQL 内置的全文搜索相结合，当用户搜索嵌入向量容易模糊处理的确切标识符、错误代码或产品名称时，这很有帮助。而元数据过滤（对租户、日期或文档类型的 WHERE 子句）通常是多租户系统中检索质量提升最大的单一因素，因为关于哪些文本块相关的最强信号往往根本不是语义上的。

重排序

向量搜索是一种召回工具。它能够可靠地将相关文本块纳入前 20 名，但在这 20 名内部的排序是松散的，而答案质量取决于模型实际阅读的前 5 名内容。重排序器将问题与每个候选文本块放在一起查看，并根据实际相关性重新排序，而分别针对问题和文本块计算的嵌入向量无法完全判断这种相关性。

我们使用一个小型快速的 LLM 作为重排序器。DeepSeek V4 Flash 每百万输入 token 成本为 0.068 美元，并且能很好地处理列表式排序：

import json

RERANK_PROMPT = """You are ranking text passages by relevance to a question.

Question: {question}

Passages:
{passages}

Return a JSON array of the {n} passage numbers most relevant to the
question, most relevant first. Output only the JSON array."""

def rerank(question: str, candidates: list[dict], top_n: int = 5) -> list[dict]:
    passages = "\n\n".join(
        f"[{i}] {c['text']}" for i, c in enumerate(candidates)
    )
    resp = client.chat.completions.create(
        model="deepseek-v4-flash",
        messages=[{
            "role": "user",
            "content": RERANK_PROMPT.format(
                question=question, passages=passages, n=top_n
            ),
        }],
        max_tokens=64,
        temperature=0,
    )
    try:
        order = json.loads(resp.choices[0].message.content)
        return [candidates[i] for i in order[:top_n] if i < len(candidates)]
    except (json.JSONDecodeError, TypeError):
        return candidates[:top_n]   # fall back to vector order

注意这个回退机制：如果重排序器返回了格式错误的JSON，我们会提供向量搜索顺序的结果，而不是让请求失败。重排序器是为了改进答案，它们绝不应成为故障点。

这一步就是成本模型标记为“隐性支出”的环节。每个查询大约读取10,400个输入token，是答案模型读取量的四倍，在低流量时微不足道，在高流量时却是最大的开支项目。成本部分提到的调整手段（减少候选数量、条件性重排序或托管重排序器）在这里都适用。

生成答案

最后一步将排名靠前的文本块组装成提示词，要求模型严格依据这些内容来作答，使用gpt-oss-120b，原因在成本模型中已经说明：

SYSTEM_PROMPT = """Answer the user's question using only the provided context.
Cite the source of each claim using the [doc_id] shown with each passage.
If the context does not contain the answer, say so plainly."""

def answer(question: str) -> str:
    candidates = retrieve(question, k=20)
    top = rerank(question, candidates, top_n=5)
    context = "\n\n".join(f"[{c['doc_id']}] {c['text']}" for c in top)
    resp = client.chat.completions.create(
        model="openai-gpt-oss-120b",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": f"Context:\n{context}\n\nQuestion: {question}"},
        ],
        max_tokens=512,
        temperature=0.2,
    )
    return resp.choices[0].message.content

将这个包装在你选择的Web框架中；一个启用了流式传输的FastAPI端点（stream=True）是常见形态，流式传输对感知延迟很重要，因为用户在其他部分生成时就能看到最初的单词。

提示结构正在发挥实际作用。指示模型仅根据上下文回答，并在上下文不足时承认，这是你对抗自信捏造的主要防线，而将doc_id传递到提示中使答案可引用。用户会原谅“我没有那方面的信息。”他们不会原谅被归因于他们自己文档的虚构政策细节。

这条流水线中的所有内容也可以作为托管服务获得：DigitalOcean 知识库在单个检索端点后面处理分块、嵌入、OpenSearch支持的向量存储和重排序，按索引和检索的令牌数计费，再加上OpenSearch集群成本（每月19美元起）。DIY流水线让你完全控制分块、混合搜索、过滤和重排序策略，并将向量保存在一个可以用纯SQL查询的数据库中。托管路径能让你更快地获得一个可运行的代理。两者使用相同的嵌入模型，价格相同，所以上述成本模型可以套用。

本文最初提出的论点经得起逐项数字的检验。RAG作为一种昂贵架构的名声源于错误的思维模式：团队在应该为查询定价时却为语料库定价。嵌入100,000个文档的成本约为7美元，存储向量每月约1美元，而重排序和生成占模型支出的99%以上，随着流量和每次查询读取的令牌数线性扩展（它们从每天1,000次查询时占总账单的38%——此时固定数据库成本占主导——增长到100,000次时的94%）。

控制成本的三个决策依次是：哪个模型生成答案（4倍波动），每次查询读取多少个分块（重排序乘数），以及何时将热工作负载从无服务器迁移到专用容量（每月约3,200美元的令牌支出）。这些都不是奇特的优化；它们都可见于你可以在电子表格中构建的成本模型，本文为你提供了构建自己模型的公式。

• DigitalOcean Inference定价
• 生成向量嵌入
• 在DigitalOcean上推出统一批量推理以及批量推理指南
• 最佳OpenAI兼容推理API：2026年的即插即用替代方案
• 托管PostgreSQL定价和支持的扩展
• 无服务器推理上的提示缓存
• 知识库和分块最佳实践
• pgvector