RAG 是什么，为什么需要

"""三条路怎么选：RAG、Function Call、微调的路由打分。

不写死"什么场景一定用什么"，而是把判断拆成几个可观察的信号，
每个信号给三条路各自加权。换了业务只需要加信号，不用改结构。

纯标准库，直接运行可看到六个真实需求的判定过程。
"""

# 每个信号是一个 (问题, 三条路的加权) 三元组。
# 权重是经验值，重点在"哪条路被这个信号支持"，不在数字本身。
SIGNALS = [
    ("答案写在某份文档里，能指出是哪一段吗", {"rag": 3, "fc": 0, "ft": -1}),
    ("资料是否经常更新（周级或更快）", {"rag": 3, "fc": 1, "ft": -3}),
    ("需要回答时给出出处，让人能去核对", {"rag": 3, "fc": 0, "ft": -2}),
    ("答案要取自实时状态（库存、订单、天气）", {"rag": -1, "fc": 3, "ft": -2}),
    ("要让模型真的去做一个动作（下单、发消息、写库）", {"rag": -2, "fc": 3, "ft": -1}),
    ("答案需要一次精确计算而不是复述", {"rag": -1, "fc": 3, "ft": 0}),
    ("要改变的是说话风格、术语习惯或输出格式", {"rag": 0, "fc": 0, "ft": 3}),
    ("要让模型掌握一种它没见过的任务范式", {"rag": 0, "fc": 0, "ft": 3}),
    ("有数千条以上标注好的样本可用", {"rag": 0, "fc": 0, "ft": 2}),
    ("推理延迟和单条成本必须压到最低", {"rag": -1, "fc": -1, "ft": 2}),
]

NAMES = {"rag": "RAG 检索增强", "fc": "Function Call 函数调用", "ft": "微调 Fine-tune"}


def decide(answers):
    """answers: 与 SIGNALS 等长的布尔列表。返回按得分排序的三条路。"""
    score = {"rag": 0, "fc": 0, "ft": 0}
    for hit, (_q, w) in zip(answers, SIGNALS):
        if hit:
            for k in score:
                score[k] += w[k]
    return sorted(score.items(), key=lambda kv: kv[1], reverse=True)


CASES = [
    ("内部制度问答：员工问年假怎么休",
     [1, 1, 1, 0, 0, 0, 0, 0, 0, 0]),
    ("查某个订单现在到哪了",
     [0, 0, 0, 1, 0, 0, 0, 0, 0, 0]),
    ("让模型按公司模板写周报",
     [0, 0, 0, 0, 0, 0, 1, 0, 1, 1]),
    ("客服问答：既要查产品手册，又要查订单状态",
     [1, 1, 1, 1, 0, 0, 0, 0, 0, 0]),
    ("把长篇法条改写成客户听得懂的话，并注明依据",
     [1, 0, 1, 0, 0, 0, 1, 0, 0, 0]),
    ("医疗问诊：需要专业语感 + 严格的问答格式",
     [0, 0, 0, 0, 0, 0, 1, 1, 1, 0]),
]


def main():
    for title, answers in CASES:
        ranked = decide(answers)
        top, second = ranked[0], ranked[1]
        print("需求：%s" % title)
        for key, val in ranked:
            bar = "█" * max(val, 0)
            print("    %-24s %3d %s" % (NAMES[key], val, bar))
        # 两条路分差小于 2 时，实践中多半是"要一起上"，而不是二选一
        verdict = NAMES[top[0]]
        if top[1] - second[1] < 2:
            verdict = "%s + %s（组合）" % (NAMES[top[0]], NAMES[second[0]])
        print("    → 建议：%s\n" % verdict)


if __name__ == "__main__":
    main()

"""算一笔账：每次提问都把整本资料塞进提示词，代价是多少。

"既然模型的上下文窗口越来越大，为什么不直接把所有文档都塞进去？"
这是学 RAG 时最常见的疑问。答案不是"塞不下"，而是三笔账都划不来：
钱、延迟、准确率。这个脚本把前两笔算成具体数字。

纯标准库，直接 python3 context_cost.py 就能跑。
"""

# 中文按字计 token 偏保守：一个汉字通常落在 1 个 token 上下，
# 这里取 1.0，估出来的是量级而不是精确账单。
TOKENS_PER_CHINESE_CHAR = 1.0

# 一份中等规模的企业资料：500 份文档，每份平均 3000 字
DOC_COUNT = 500
CHARS_PER_DOC = 3000

# 检索方案：每次只取回 4 个块，每块 400 字
TOP_K = 4
CHARS_PER_CHUNK = 400

QUERIES_PER_DAY = 2000

# 输入价格，单位：元 / 百万 token。改成自己实际用的模型价格再跑一遍。
PRICE_PER_MTOK = 2.0


def tokens(chars):
    return chars * TOKENS_PER_CHINESE_CHAR


def yuan(tok):
    return tok / 1_000_000 * PRICE_PER_MTOK


def main():
    full_chars = DOC_COUNT * CHARS_PER_DOC
    rag_chars = TOP_K * CHARS_PER_CHUNK

    full_tok = tokens(full_chars)
    rag_tok = tokens(rag_chars)

    print("全量塞入：每次提问送进去 %s 字 ≈ %s token" % (f"{full_chars:,}", f"{int(full_tok):,}"))
    print("检索后送：每次提问送进去 %s 字 ≈ %s token" % (f"{rag_chars:,}", f"{int(rag_tok):,}"))
    print("倍数差距：%.0f 倍" % (full_tok / rag_tok))
    print("-" * 56)

    full_day = yuan(full_tok) * QUERIES_PER_DAY
    rag_day = yuan(rag_tok) * QUERIES_PER_DAY
    print("按 %s 次/天、%.1f 元/百万 token 估算（只算输入）：" % (f"{QUERIES_PER_DAY:,}", PRICE_PER_MTOK))
    print("  全量塞入：%10.2f 元/天　%12.2f 元/年" % (full_day, full_day * 365))
    print("  检索后送：%10.2f 元/天　%12.2f 元/年" % (rag_day, rag_day * 365))
    print("  一年省下：%10.2f 元" % ((full_day - rag_day) * 365))
    print("-" * 56)

    # 第三笔账没法用公式算，但必须知道它存在：
    # 上下文里无关内容越多，模型越容易被带偏。检索的价值不只是省钱。
    print("还有一笔算不出来的账：")
    print("  送进去 %s token，其中真正相关的只有 %s token，" % (f"{int(full_tok):,}", f"{int(rag_tok):,}"))
    print("  剩下 %.2f%% 全是干扰项。" % ((1 - rag_tok / full_tok) * 100))
    print("  上下文越长，模型越容易漏掉夹在中间的关键句——这是省钱之外的主因。")

    # 上下文窗口能不能装得下，也顺手验一下
    for window in (32_000, 128_000, 1_000_000):
        fit = "装得下" if full_tok <= window else "装不下"
        print("  窗口 %9s token：%s" % (f"{window:,}", fit))


if __name__ == "__main__":
    main()

"""同一个问题、同一个模型，给不给 context 的差别在哪。

这个脚本不联网、不调模型，它只做一件事：把两种提示词完整打印出来，
放在一起看。RAG 改变的就是这段文本，别的什么都没改。

顺带演示三条必须写进模板的纪律：
  1. 明确圈定"只能用已知信息"
  2. 给出兜底话术，让模型有"不知道"可说
  3. 让模型带上出处编号，答案才可核对
"""

QUESTION = "货物 ABC123456 现在到哪了？预计什么时候到？"

CHUNKS = [
    ("物流信息.pdf#p1", "货物编号：ABC123456，发货日期：2023-01-15，当前位置：上海分拨中心。"),
    ("物流信息.pdf#p1", "预计到达日期：2023-01-20。"),
]

NAKED = """{question}"""

GROUNDED = """你是物流信息查询助手。请只根据【已知信息】回答【问题】。

规则：
1. 只使用【已知信息】里出现过的内容，不允许补充任何外部知识。
2. 【已知信息】不足以回答时，回答"资料中未提及"，不要推测。
3. 答案末尾用方括号列出你用到的信息来源编号。

【已知信息】
{context}

【问题】
{question}"""


def build_context(chunks):
    """每块前面加编号，模型才有东西可引用。

    只把正文拼成一坨，模型就算想标出处也无从标起——
    "让它给出处"这件事一半以上是在这一步实现的，不是靠提示词多说一句。
    """
    lines = []
    for i, (source, text) in enumerate(chunks, 1):
        lines.append("[%d] （来源：%s）%s" % (i, source, text))
    return "\n".join(lines)


def main():
    print("=" * 62)
    print("A. 不做 RAG：模型手上只有问题")
    print("=" * 62)
    print(NAKED.format(question=QUESTION))
    print()
    print("模型此时的处境：它从没见过这份运单。")
    print("它要么说不知道，要么按训练数据里见过的运单格式编一个日期出来。")
    print("这就是幻觉——不是模型坏，是它被要求回答一个手上没有依据的问题。")

    print()
    print("=" * 62)
    print("B. 做了 RAG：同一个问题，前面多了两段检索回来的原文")
    print("=" * 62)
    prompt = GROUNDED.format(context=build_context(CHUNKS), question=QUESTION)
    print(prompt)

    print()
    print("-" * 62)
    print("提示词长度：无 context %d 字符 → 有 context %d 字符"
          % (len(NAKED.format(question=QUESTION)), len(prompt)))
    print("多出来的部分全是原文和规则，没有一个字是模型自己想出来的。")

    # 把检索结果换成空的，看模板还撑不撑得住
    print()
    print("=" * 62)
    print("C. 检索什么都没找到：模板必须能兜住")
    print("=" * 62)
    empty = GROUNDED.format(context="（无）", question="公司年假怎么休？")
    print(empty)
    print()
    print("规则 2 在这里起作用：模型应当回答「资料中未提及」。")
    print("没有这条规则，空 context 反而最容易触发编造。")


if __name__ == "__main__":
    main()

"""最小 RAG 内核：不装任何第三方库，把五个步骤跑一遍。

这份文件的目的不是做一个好用的检索器，而是证明 RAG 没有魔法：
向量化 = 把文字变成一串数字，检索 = 比较两串数字的夹角，
生成 = 把找回来的文字拼进一段提示词。

这里用最朴素的词袋（bag of words）当作向量化，
真实项目里换成 Embedding 模型即可，其余四步一个字都不用改。
"""
import math
import re
from collections import Counter

# 知识库：每一条就是一个 chunk（切好的一块文本）
CHUNKS = [
    "速达物流公司总部在北京市，业务范围是国际快递和仓储管理。",
    "货物编号 ABC123456 的发货日期是 2023-01-15，当前位置在上海分拨中心。",
    "该批货物的预计到达日期是 2023-01-20。",
    "运输公司是快运通，运输方式为陆运，出发地广州，目的地重庆，预计运输时间 3 天。",
    "仓库名称是东方仓储中心，位于深圳市，存储电子产品，常温仓储，当前库存 1000 件。",
]


def tokenize(text):
    """中文按字切，英文数字按词切。

    中文没有空格，直接按空格切会把整句话当成一个词，
    词袋里只剩一个维度，任何两句话的相似度都会变成 0 或 1。
    """
    text = text.lower()
    tokens = re.findall(r"[a-z0-9]+", text)          # 英文与数字整体保留
    tokens += re.findall(r"[\u4e00-\u9fff]", text)   # 汉字逐字拆
    return tokens


def embed(text):
    """把一段文字变成 {词: 次数} 的稀疏向量。

    换成真正的 Embedding 模型时，这里返回的是一个定长浮点数列表，
    下面的 cosine 一行都不用改——它只依赖"向量"这个抽象。
    """
    return Counter(tokenize(text))


def cosine(v1, v2):
    """余弦相似度：两个向量夹角的余弦值，范围 [-1, 1]，词袋下恒为 [0, 1]。"""
    common = set(v1) & set(v2)
    dot = sum(v1[k] * v2[k] for k in common)
    n1 = math.sqrt(sum(x * x for x in v1.values()))
    n2 = math.sqrt(sum(x * x for x in v2.values()))
    if n1 == 0 or n2 == 0:
        return 0.0
    return dot / (n1 * n2)


def build_index(chunks):
    """离线阶段：把每个 chunk 算成向量存起来。"""
    return [(c, embed(c)) for c in chunks]


def retrieve(index, question, k=2):
    """在线阶段：问题也算成向量，按相似度排序取前 k 个。"""
    qv = embed(question)
    scored = [(cosine(qv, cv), c) for c, cv in index]
    scored.sort(key=lambda x: x[0], reverse=True)
    return scored[:k]


PROMPT_TEMPLATE = """基于以下已知信息，简洁、专业地回答用户问题。
已知信息里没有的内容一律回答"资料中未提及"，不允许编造。

已知信息:
{context}

问题:
{question}"""


def build_prompt(hits, question):
    context = "\n".join(text for _score, text in hits)
    return PROMPT_TEMPLATE.format(context=context, question=question)


if __name__ == "__main__":
    index = build_index(CHUNKS)
    question = "我的快递出发地是哪？预计几天到？"

    hits = retrieve(index, question, k=2)
    for score, text in hits:
        print("%.4f  %s" % (score, text))

    print("-" * 60)
    print(build_prompt(hits, question))

    # 库里根本没有的问题：检索照样会返回分数最高的两条，
    # 挡住编造的是提示词里那句"未提及"，不是检索本身。
    print("-" * 60)
    miss = retrieve(index, "公司的年假制度是怎么规定的？", k=2)
    for score, text in miss:
        print("%.4f  %s" % (score, text))

身高：160-170cm， 体重：90-115斤，建议尺码M。
身高：165-175cm， 体重：115-135斤，建议尺码L。
身高：170-178cm， 体重：130-150斤，建议尺码XL。
身高：175-182cm， 体重：145-165斤，建议尺码2XL。
身高：178-185cm， 体重：160-180斤，建议尺码3XL。
身高：180-190cm， 体重：180-210斤，建议尺码4XL。
面料分类：其他
图案：纯色
领型：翻领
衣门襟：单排扣
颜色：黑色 卡其色 粉色 杏色
袖型：收口袖
适用季节：冬季
袖长：长袖
厚薄：厚款
适用场景：其他休闲
衣长：常规款
版型：宽松型
款式细节：假两件
工艺处理：免烫处理
适用对象：青年
面料功能：保暖
穿搭方式：外穿
销售渠道类型：纯电商(只在线上销售)
材质成分：棉100%

"""完整案例：把一份商品属性表变成可问答的知识库。

场景很小，但五步一步不少：加载 → 切分 → 向量化 → 检索 → 拼提示词。
数据就是一份普通的商品属性文本，读者换成自己的文件即可复用。

纯标准库，可以直接运行。真实项目里把 embed() 换成 Embedding 模型、
把线性扫描换成向量库，其余逻辑原封不动。
"""
import math
import os
import re
from collections import Counter

HERE = os.path.dirname(os.path.abspath(__file__))
KB_FILE = os.path.join(HERE, "size_kb.txt")


def load(path):
    """第一步：加载。真实项目里这里换成 PDF / Word / 网页加载器。"""
    with open(path, encoding="utf-8") as f:
        return f.read()


def split_by_line(text):
    """第二步：切分。

    这份资料每一行就是一条独立事实，行本身就是天然的语义边界，
    按行切比按固定字数切更好——切分规则应当顺着资料的结构走，
    而不是不管什么资料都套一个 chunk_size。
    """
    return [ln.strip() for ln in text.splitlines() if ln.strip()]


def tokenize(text):
    text = text.lower()
    return re.findall(r"[a-z0-9]+", text) + re.findall(r"[\u4e00-\u9fff]", text)


def embed(text):
    """第三步：向量化（此处用词袋代替 Embedding 模型）。"""
    return Counter(tokenize(text))


def cosine(a, b):
    common = set(a) & set(b)
    dot = sum(a[k] * b[k] for k in common)
    na = math.sqrt(sum(v * v for v in a.values()))
    nb = math.sqrt(sum(v * v for v in b.values()))
    return dot / (na * nb) if na and nb else 0.0


class MiniStore:
    """第三步的落点：一个最小向量库，只有 add 和 search 两个动作。"""

    def __init__(self):
        self.items = []

    def add(self, chunks):
        for c in chunks:
            self.items.append((c, embed(c)))

    def search(self, query, k=3, score_min=0.0):
        qv = embed(query)
        hits = [(cosine(qv, v), c) for c, v in self.items]
        hits.sort(key=lambda x: x[0], reverse=True)
        return [h for h in hits[:k] if h[0] >= score_min]


TEMPLATE = """你是导购助手，只根据【已知信息】回答顾客问题。
已知信息里没有的，回答"这个资料里没写"，不要猜。

【已知信息】
{context}

【顾客问题】
{question}"""


def answer(store, question, k=3, score_min=0.05):
    """第四、五步：检索 + 拼提示词。"""
    hits = store.search(question, k=k, score_min=score_min)
    if not hits:
        return None, "检索没有命中任何内容，直接回复顾客：这个资料里没写。"
    context = "\n".join("- %s" % text for _s, text in hits)
    return hits, TEMPLATE.format(context=context, question=question)


def main():
    raw = load(KB_FILE)
    chunks = split_by_line(raw)
    print("原文 %d 字符 → 切成 %d 块" % (len(raw), len(chunks)))

    store = MiniStore()
    store.add(chunks)
    print("入库完成，库里 %d 条向量\n" % len(store.items))

    questions = [
        "我身高175体重140，穿多大码？",
        "这件衣服是什么材质的？",
        "有没有红色的？",
        "这件衣服适合夏天穿吗？",
    ]

    for q in questions:
        print("=" * 60)
        print("问：%s" % q)
        hits, prompt = answer(store, q)
        if hits is None:
            print(prompt)
            continue
        for score, text in hits:
            print("   %.4f  %s" % (score, text))
        print("-" * 60)
        print(prompt)
        print()


if __name__ == "__main__":
    main()

"""RAG 骨架模板：把 TODO 填掉就是一个可用的知识库问答。

用法
    1. 把 DOC_DIR 指向自己的资料目录
    2. 按注释换掉 embed / VectorStore / call_llm 三处实现
    3. python3 rag_skeleton_min.py "你的问题"

结构刻意分成五个函数，和 RAG 的五步一一对应。
换向量库、换 Embedding、换模型，改的都只是函数体，主流程不动。
"""
import math
import os
import re
import sys
from collections import Counter

# ---------------------------------------------------------------- 配置
DOC_DIR = os.environ.get("RAG_DOC_DIR", "./docs")   # TODO: 换成自己的资料目录
CHUNK_SIZE = 300        # TODO: 按资料形态调，见「切分」一节
CHUNK_OVERLAP = 50      # TODO: 一般取 chunk_size 的 10%~20%
TOP_K = 4               # TODO: 先从 3~5 试起
SCORE_MIN = 0.0         # TODO: 调完 k 之后再调这个，不要两个一起动


# ------------------------------------------------------- ① 加载
def load_docs(doc_dir):
    """返回 [(来源标识, 全文), ...]。

    TODO: 需要读 PDF / Word / 网页时，在这里接对应的加载器；
    加载器只负责"拿到纯文本 + 记住它从哪来"这两件事。
    """
    docs = []
    if not os.path.isdir(doc_dir):
        return docs
    for name in sorted(os.listdir(doc_dir)):
        path = os.path.join(doc_dir, name)
        if os.path.isfile(path) and name.lower().endswith((".txt", ".md")):
            with open(path, encoding="utf-8") as f:
                docs.append((name, f.read()))
    return docs


# ------------------------------------------------------- ② 切分
def split(text, size=CHUNK_SIZE, overlap=CHUNK_OVERLAP):
    """按段落聚合，超长再按字数硬切，块间保留 overlap 个字符。"""
    parts, buf = [], ""
    for para in re.split(r"\n\s*\n", text):
        para = para.strip()
        if not para:
            continue
        if len(buf) + len(para) <= size:
            buf = (buf + "\n" + para).strip()
        else:
            if buf:
                parts.append(buf)
            while len(para) > size:
                parts.append(para[:size])
                para = para[size - overlap:]
            buf = para
    if buf:
        parts.append(buf)
    return parts


# ------------------------------------------------------- ③ 向量化
def embed(text):
    """TODO: 换成真正的 Embedding 模型，返回定长浮点数列表。

    换实现时只需保证：入参是字符串，返回值能被 similarity() 处理。
    ⚠️ 建库与提问必须用同一个模型，换模型就要整库重建。
    """
    tokens = re.findall(r"[a-z0-9]+", text.lower()) + re.findall(r"[\u4e00-\u9fff]", text)
    return Counter(tokens)


def similarity(a, b):
    """余弦相似度。换成浮点向量后改成标准点积除模长即可。"""
    common = set(a) & set(b)
    dot = sum(a[k] * b[k] for k in common)
    na = math.sqrt(sum(v * v for v in a.values()))
    nb = math.sqrt(sum(v * v for v in b.values()))
    return dot / (na * nb) if na and nb else 0.0


# ------------------------------------------------------- ④ 存与检索
class VectorStore:
    """TODO: 上生产时换成向量库；接口保持 add / search 两个方法不变。"""

    def __init__(self):
        self.rows = []

    def add(self, source, chunk):
        self.rows.append({"source": source, "text": chunk, "vec": embed(chunk)})

    def search(self, query, k=TOP_K, score_min=SCORE_MIN):
        qv = embed(query)
        scored = [(similarity(qv, r["vec"]), r) for r in self.rows]
        scored.sort(key=lambda x: x[0], reverse=True)
        return [(s, r) for s, r in scored[:k] if s >= score_min]


# ------------------------------------------------------- ⑤ 生成
PROMPT = """请只根据【已知信息】回答【问题】。
【已知信息】里没有的内容，回答"资料中未提及"，不要推测。
答案末尾用方括号标出用到的编号。

【已知信息】
{context}

【问题】
{question}"""


def build_prompt(hits, question):
    lines = ["[%d] （%s）%s" % (i, r["source"], r["text"])
             for i, (_s, r) in enumerate(hits, 1)]
    return PROMPT.format(context="\n".join(lines) or "（无）", question=question)


def call_llm(prompt):
    """TODO: 换成真实的模型调用，密钥一律走环境变量。

        api_key = os.environ.get("LLM_API_KEY")
        ...
    """
    return "[未接入模型] 下面是本应发给模型的提示词：\n" + prompt


def main():
    question = sys.argv[1] if len(sys.argv) > 1 else "TODO: 在这里写一个测试问题"

    store = VectorStore()
    total = 0
    for source, text in load_docs(DOC_DIR):
        for chunk in split(text):
            store.add(source, chunk)
            total += 1
    print("入库 %d 块（来自 %s）" % (total, DOC_DIR))

    hits = store.search(question)
    for score, row in hits:
        print("  %.4f  %s  %s" % (score, row["source"], row["text"][:40].replace("\n", " ")))

    print("-" * 60)
    print(call_llm(build_prompt(hits, question)))


if __name__ == "__main__":
    main()