1
# 步骤 1：调 Embedding 模型把问题变成向量
2
vec = get_embedding(query)
3

4
# 步骤 2：调 ChromaDB 搜相似片段
5
results = collection.query(query_embeddings=[vec], n_results=3)
6

7
# 步骤 3：自己拼 System Prompt（f-string）
8
system_prompt = f"根据以下资料回答：{chunks}"
9

10
# 步骤 4：调 LLM API
11
response = client.chat.completions.create(
12
    model="...", messages=[{"role": "system", ...}, {"role": "user", ...}]
13
)
14

15
# 步骤 5：处理流式输出
16
for chunk in response:
17
    yield chunk.choices[0].delta.content
18

19
# 步骤 6：自己算 Token、做截断、防幻觉...

1
rag_chain = (
2
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
3
    | prompt
4
    | llm
5
    | StrOutputParser()
6
)
7
# 一行代码 = 上面 6 个步骤
8
result = rag_chain.invoke("问题")

1
取餐盘 → 盛米饭 → 打菜 → 浇汤 → 端走
2
   |        |       |      |      |
3
   └────────┴───────┴──────┴──────┘
4
         每个师傅只做一件事
5
         做完传给下一个师傅

1
step1_out = do_step1(input_data)
2
step2_out = do_step2(step1_out)
3
step3_out = do_step3(step2_out)
4
result = do_step4(step3_out)

1
chain = do_step1 | do_step2 | do_step3 | do_step4
2
result = chain.invoke(input_data)

1
from langchain_core.documents import Document
2

3
doc = Document(
4
    page_content="苹果富含维生素C，每天吃一个有益健康。",  # 正面：内容
5
    metadata={"source": "营养百科", "page": 12}              # 背面：来源信息
6
)

1
# retriever 返回的就是 List[Document]
2
docs = retriever.invoke("苹果怎么吃？")
3
for doc in docs:
4
    print(doc.page_content)  # 片段内容
5
    print(doc.metadata)      # {title: "...", source: "..."}

1
from langchain_chroma import Chroma
2
import chromadb
3

4
# 连接到现有的 ChromaDB（和手搓版共用同一个数据库）
5
chroma_client = chromadb.PersistentClient(path="./chroma_db")
6

7
vectorstore = Chroma(
8
    client=chroma_client,              # 复用已有的客户端
9
    collection_name="documents",       # 集合名（手搓版也是这个）
10
    embedding_function=embeddings,     # 告诉它：查询时用哪个翻译官
11
)
12

13
# 统计有多少条文档
14
count = vectorstore._collection.count()

1
# 从 VectorStore 创建检索器
2
retriever = vectorstore.as_retriever(
3
    search_type="similarity",       # 检索方式：按相似度
4
    search_kwargs={"k": 3},         # 返回前 3 个最相关的结果
5
)
6

7
# 使用检索器
8
docs = retriever.invoke("苹果怎么吃？")
9
# 返回 List[Document]，已按相似度排好序

1
# 手搓版：直接调 Chroma 的 query API
2
results = collection.query(
3
    query_embeddings=[vec],
4
    n_results=3,
5
    include=["documents", "metadatas", "distances"]
6
)
7
# 返回的是原始字典，需要手动解析
8

9
# LangChain 版：retriever 帮你封装了上述所有步骤
10
# 输入字符串 → 自动调 Embedding → 自动查 VectorStore → 返回 Document 对象
11
# 你只需要关心：输入问题，拿到文档

1
# ========== 第 1 步：定义 Prompt 模板 ==========
2
RAG_PROMPT = ChatPromptTemplate.from_messages([
3
    ("system", "根据以下资料回答：\n\n{context}"),
4
    ("human", "{question}"),
5
])
6

7
# ========== 第 2 步：组装链 ==========
8
rag_chain = (
9
    {
10
        "context": retriever | format_docs,   # 检索 → 格式化 → 填入 {context}
11
        "question": RunnablePassthrough(),     # 用户问题原样填入 {question}
12
    }
13
    | RAG_PROMPT    # 把字典里的值填进 Prompt 模板
14
    | llm           # 把 messages 送给 LLM
15
    | StrOutputParser()  # 把 AIMessage 转成纯字符串
16
)

1
用户输入: "苹果怎么吃？"
2
    |
3
    v
4
+------------------+
5
| RunnablePassthrough |  ← 原样通过
6
|   (question)       |
7
+------------------+
8
    |
9
    v
10
+------------------+
11
|    retriever     |  ← Embedding → ChromaDB 搜相似文档
12
+------------------+
13
    |
14
    v
15
+------------------+
16
|   format_docs    |  ← List[Document] 拼成字符串
17
|   (context)      |
18
+------------------+
19
    |
20
    v
21
+------------------+
22
|  RAG_PROMPT      |  ← {context} + {question} 填入模板
23
+------------------+
24
    |
25
    v
26
+------------------+
27
|      llm         |  ← 调 DeepSeek 生成回答
28
+------------------+
29
    |
30
    v
31
+------------------+
32
| StrOutputParser  |  ← 转纯字符串
33
+------------------+
34
    |
35
    v
36
最终输出: "根据资料，苹果可以生吃..."

1
{
2
    "context": retriever | format_docs,    # 这条路需要处理（检索+格式化）
3
    "question": RunnablePassthrough(),      # 这条路不需要处理（原样通过）
4
}

1
# 没有 StrOutputParser
2
result = (prompt | llm).invoke("你好")
3
print(result)
4
# 输出: AIMessage(content='你好！', additional_kwargs={}, ...)
5

6
# 有 StrOutputParser
7
result = (prompt | llm | StrOutputParser()).invoke("你好")
8
print(result)
9
# 输出: "你好！"

1
尊敬的{老师}：
2
    我是{班级}的{姓名}，因{原因}请假{天数}天。
3
                                        申请人：{姓名}

1
RAG_PROMPT = ChatPromptTemplate.from_messages([
2
    ("system", "你是助手，根据资料回答。资料：\n{context}"),
3
    ("human", "{question}"),
4
])
5

6
# 运行时自动填入
7
messages = RAG_PROMPT.format_messages(
8
    context="苹果可以生吃...",
9
    question="苹果怎么吃？"
10
)
11
# 生成:
12
# [SystemMessage(content="你是助手..."), HumanMessage(content="苹果怎么吃？")]

1
# 手搓版：手动拼字符串，容易漏、容易错
2
messages = [
3
    {"role": "system", "content": f"根据资料回答：{context}"},
4
    {"role": "user", "content": query}
5
]
6

7
# LangChain 版：模板化管理，可复用、可维护
8
messages = RAG_PROMPT.format_messages(context=context, question=query)

1
# 同步一次性调用
2
result = rag_chain.invoke("苹果怎么吃？")
3
print(result)  # 等全部生成完才打印
4

5
# 异步流式调用（咱们代码里用的）
6
async for chunk in rag_chain.astream("苹果怎么吃？"):
7
    print(chunk, end="")  # 生成一个字打印一个字

1
# LangChain 输出: "苹" → "果" → "可" → "以" → ...
2

3
# SSE 格式: data: {"content": "苹"}\n\n
4
data: {"content": "果"}\n\n

1
async for chunk in rag_chain.astream(req.query):
2
    data = {"type": "content", "content": chunk}
3
    yield f"data: {json.dumps(data, ensure_ascii=False)}\n\n"

1
llm = ChatDeepSeek(
2
    model="deepseek-chat",
3
    api_key="你的密钥",
4
    streaming=True,
5
    # 关键：开启推理链
6
    model_kwargs={
7
        "extra_body": {
8
            "thinking": {"type": "enabled"}
9
        }
10
    },
11
)

1
async for chunk in llm.astream("问题"):
2
    # chunk 是 AIMessageChunk 对象
3

4
    # 提取推理过程（草稿纸内容）
5
    reasoning = chunk.additional_kwargs.get("reasoning_content", "")
6
    if reasoning:
7
        print(f"[思考] {reasoning}")
8

9
    # 提取正式回答（答题卡内容）
10
    content = chunk.content
11
    if content:
12
        print(f"[回答] {content}")

1
from dotenv import load_dotenv
2
import os
3

4
# 加载 .env 文件（必须在代码最前面执行一次）
5
load_dotenv()
6

7
# 读取配置
8
LLM_MODEL = os.getenv("LLM_MODEL", "deepseek-chat")
9
EMBEDDING_MODEL = os.getenv("EMBEDDING_MODEL", "BAAI/bge-small-zh")
10
CHROMA_DB_PATH = os.getenv("CHROMA_DB_PATH", "./chroma_db")

1
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
2
from langchain_core.output_parsers import StrOutputParser
3
from langchain_core.prompts import ChatPromptTemplate
4

5
# 透明通过
6
RunnablePassthrough()           # 原样输出
7
RunnablePassthrough.assign(...) # 原样通过，同时附加新字段
8

9
# 自定义函数包装
10
RunnableLambda(lambda x: x.upper())  # 把普通函数变成链的一环
11

12
# 输出解析
13
StrOutputParser()               # 转字符串（会丢掉额外信息）
14
JsonOutputParser()              # 转 JSON

1
# 同步
2
result = chain.invoke("输入")
3

4
# 异步
5
result = await chain.ainvoke("输入")
6

7
# 批量
8
results = chain.batch(["输入1", "输入2", "输入3"])
9

10
# 流式（同步）
11
for chunk in chain.stream("输入"):
12
    print(chunk)
13

14
# 流式（异步）
15
async for chunk in chain.astream("输入"):
16
    print(chunk)

1
# 方式 1：from_messages（推荐，对应 Chat 模型）
2
prompt = ChatPromptTemplate.from_messages([
3
    ("system", "你是{role}，擅长{skill}"),
4
    ("human", "{question}"),
5
])
6

7
# 方式 2：from_template（简单场景）
8
prompt = ChatPromptTemplate.from_template("请回答：{question}")

1
from langchain_huggingface import HuggingFaceBgeEmbeddings
2
import torch
3

4
device = "mps" if torch.backends.mps.is_available() else "cpu"
5

6
embeddings = HuggingFaceBgeEmbeddings(
7
    model_name="BAAI/bge-small-zh",
8
    model_kwargs={"device": device},
9
    encode_kwargs={"normalize_embeddings": True},
10
)

1
from langchain_deepseek import ChatDeepSeek
2

3
llm = ChatDeepSeek(
4
    model="deepseek-chat",
5
    api_key="你的密钥",
6
    streaming=True,
7
    model_kwargs={
8
        "extra_body": {
9
            "thinking": {"type": "enabled"}  # 开启推理链
10
        }
11
    },
12
)
13

14
# 流式提取推理链
15
async for chunk in llm.astream("问题"):
16
    reasoning = chunk.additional_kwargs.get("reasoning_content", "")
17
    content = chunk.content

1
# 代码开头
2
from dotenv import load_dotenv
3
import os
4

5
load_dotenv()
6

7
LLM_MODEL = os.getenv("LLM_MODEL", "deepseek-chat")