---
  title: RAG开发
  date: 2026-06-245
---

# 一、工作流程
```text
原始文档 / 数据
        ↓
【文档收集与清洗】
        ↓
【文档切分（Chunking）】
        ↓
【向量化（Embedding）】
        ↓
【向量数据库存储】
        ↓
用户自然语言问题
        ↓
【检索系统】召回相关知识片段
        ↓
【上下文组装】构建增强 Prompt
        ↓
【大模型】基于知识生成回答
        ↓
【引用 / 校验】
        ↓
最终返回给用户（答案 + 来源）
``` 

# 二、实战
## 2.1 资料处理
### 2.1.1 pdf处理
  [PaddleOCR](https://aistudio.baidu.com/paddleocr)是百度飞桨（PaddlePaddle）团队开源的产业级 OCR（光学字符识别）与文档智能开发套件。   
```python
JOB_URL = "https://paddleocr.aistudio-app.com/api/v2/ocr/jobs"
TOKEN = ""
MODEL = "PP-OCRv5"

FILE_URL = ""

HEADERS = {
    "Authorization": f"bearer {TOKEN}",
    "Content-Type": "application/json",
}

# ========== 提交 OCR 任务 ==========
payload = {
    "fileUrl": FILE_URL,
    "model": MODEL,
    "optionalPayload": {
        "useDocOrientationClassify": False,
        "useDocUnwarping": False,
        "useChartRecognition": False,
    },
}

print("🚀 Submitting OCR job...")
resp = requests.post(JOB_URL, json=payload, headers=HEADERS)
resp.raise_for_status()

job_id = resp.json()["data"]["jobId"]
print(f"✅ Job submitted, jobId={job_id}")

# ========== 轮询任务状态 ==========
while True:
    r = requests.get(f"{JOB_URL}/{job_id}", headers=HEADERS)
    r.raise_for_status()

    data = r.json()["data"]
    state = data["state"]

    if state == "pending":
        print("⏳ Job status: pending")
    elif state == "running":
        prog = data.get("extractProgress", {})
        total = prog.get("totalPages")
        done = prog.get("extractedPages")
        if total and done:
            print(f"📄 Processing: {done}/{total} pages")
        else:
            print("📄 Processing...")
    elif state == "done":
        prog = data["extractProgress"]
        print(
            f"✅ Job finished | Pages: {prog['extractedPages']} | "
            f"Start: {prog['startTime']} | End: {prog['endTime']}"
        )
        break
    elif state == "failed":
        print("❌ Job failed:", data.get("errorMsg", "Unknown error"))
        sys.exit(1)

    time.sleep(5)

# ========== 获取 OCR 结果 ==========
jsonl_url = data["resultUrl"]["jsonUrl"]
print(f"📥 Fetching result: {jsonl_url}")

resp = requests.get(jsonl_url)
resp.raise_for_status()
```

  这里只处理了url形式的文档，实际开发中，可以搭建一个管理页面集中处理文档资料。    
  获取到的结果是按照区域划分的，还需要进一步的处理：   
```python
pages = []

for line in resp.text.strip().splitlines():
    if not line.strip():
        continue

    obj = json.loads(line)

    for page in obj.get("result", {}).get("ocrResults", []):
        pruned = page.get("prunedResult", {})
        texts = pruned.get("rec_texts", [])

        # 一页一段
        page_text = "\n".join(texts)
        pages.append(page_text)

# ✅ 最终文档
full_doc = "\n\n".join(pages)
```

## 2.2 向量化
  使用在线或者离线向量化模型处理：   
```python
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,      # 每个 chunk 最大长度
    chunk_overlap=50,    # 重叠，防止切坏语义
    separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""]
)

chunks = text_splitter.split_text(full_doc)
print(f"✅ 切分成 {len(chunks)} 个 chunk")

for i, chunk in enumerate(chunks):
    print(f"\n===== Chunk {i} =====")
    print(chunk)

model = SentenceTransformer("BAAI/bge-base-zh", cache_folder="./models")

embeddings = model.encode(
    chunks,
    normalize_embeddings=True
)

print(f"✅ 向量维度: {embeddings.shape}")

client = chromadb.PersistentClient(path="./vector_db/train_docs")
collection = client.get_or_create_collection(name="train_docs")

for i, (chunk, emb) in enumerate(zip(chunks, embeddings)):
    collection.add(
        ids=[str(i)],
        documents=[chunk],
        embeddings=[emb.tolist()]
    )

print("✅ 向量库写入完成")
```

  这里使用的是`BAAI/bge-base-zh`模型，第一次会将模型下载到本地。   
  将处理结果存储到chroma向量数据库中。   

## 2.3 检索系统
```python
model_path = "./models/models--BAAI--bge-base-zh/snapshots/0e5f83d4895db7955e4cb9ed37ab73f7ded339b6"
model = SentenceTransformer(model_path, local_files_only=True)

client = chromadb.PersistentClient(path="./vector_db/train_docs")
collection = client.get_or_create_collection(name="train_docs")

q = "培训计划怎么制定？"
q_emb = model.encode([q], normalize_embeddings=True)

res = collection.query(
    query_embeddings=q_emb.tolist(),
    n_results=3
)

print("\n\n".join(res["documents"][0]))
```

::: tip
注意这里使用离线模型时的路径
:::

## 2.4 结合大模型
  常用提示词：  
```python
你是专业的知识问答助手。
请严格基于下方提供的参考资料回答问题。
如果参考资料中不包含答案，请明确说明“当前资料无法回答该问题”。
不要编造、推测或引入外部知识。

参考资料：
"""
{{context}}
"""

用户问题：
{{question}}

请按以下要求回答：
1. 回答简洁、准确、有条理
2. 必要时使用列表或分点说明
3. 避免冗余描述
```