基于 LangChain + Milvus + 阿里云百炼构建个人博客 RAG 知识库

from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.embeddings import DashScopeEmbeddings
from pymilvus import MilvusClient, DataType

# 清洗 Markdown：去掉 YAML frontmatter、hexo 标签，转为纯文本
def extract_text_from_markdown(md_content: str) -> str:
    md_content = re.sub(r'^---\s*\n.*?---\s*\n', '', md_content, flags=re.DOTALL)
    md_content = re.sub(r'{%.*?%}', '', md_content, flags=re.DOTALL)
    html = markdown(md_content)
    soup = BeautifulSoup(html, "html.parser")
    return soup.get_text(separator="\n").strip()

# 文本切分：按标题层级优先切分，保证语义连贯
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=800,
    chunk_overlap=150,
    separators=["\n## ", "\n### ", "\n\n", "\n", "。", "；", " ", ""],
)

# 批量生成 Embedding 并插入 Milvus（每批 25 条）
for i in range(0, len(split_docs), 25):
    batch = split_docs[i:i + 25]
    vectors = embeddings.embed_documents([d.page_content for d in batch])
    client.insert(collection_name="blog_knowledge", data=[...])

RETRIEVE_TOP_K = 20
RERANK_TOP_K = 5
SIMILARITY_THRESHOLD = 0.45

def retrieve_docs(query: str):
    # 第一阶段：向量检索召回候选
    docs_with_score = vector_store.similarity_search_with_score(query, k=RETRIEVE_TOP_K)
    candidates = [(d, s) for d, s in docs_with_score if s >= SIMILARITY_THRESHOLD]
    if not candidates:
        return []
    
    # 第二阶段：Rerank 重排序
    docs = [d for d, _ in candidates]
    resp = dashscope.TextReRank.call(
        model="gte-rerank",
        query=query,
        documents=[d.page_content for d in docs],
        top_n=RERANK_TOP_K,
        return_documents=False,
    )
    
    ranked = []
    for r in resp.output.results:
        idx = r["index"]
        if r["relevance_score"] >= 0.3:
            ranked.append(docs[idx])
    return ranked

SYSTEM_PROMPT = """你是一个专业的博客知识库问答助手。你的唯一信息来源是下方提供的博客知识库片段。

【规则】
1. 你只能根据提供的知识库片段回答问题。
2. 如果知识库片段中没有相关信息，或者用户的问题与博客内容完全无关，
   你必须直接回答："这个问题与博客知识库无关，我无法回答。请尝试询问与博客内容相关的问题。"
3. 不要编造知识库片段中没有的信息。
4. 回答时要清晰、简洁、有条理。

【知识库片段】
{context}

请根据以上知识库片段回答用户问题。"""

MAX_HISTORY_TURNS = 10

def build_messages(session_id: str, user_question: str, context: str):
    session = sessions.setdefault(session_id, {"turns": [], "compressed_summary": ""})
    
    # 超过 10 轮自动压缩历史
    if len(session["turns"]) >= MAX_HISTORY_TURNS:
        summary = summarize_history(session["turns"])
        session["compressed_summary"] = summary
        session["turns"] = []
    
    messages = [{"role": "system", "content": SYSTEM_PROMPT.format(context=context)}]
    
    # 压缩摘要 + 最近对话历史
    if session["compressed_summary"]:
        messages.append({"role": "system", "content": f"此前对话摘要：{session['compressed_summary']}"})
    for turn in session["turns"]:
        messages.append({"role": "user", "content": turn["user"]})
        messages.append({"role": "assistant", "content": turn["bot"]})
    messages.append({"role": "user", "content": user_question})
    
    return messages

@app.post("/api/chat")
async def chat(request: ChatRequest):
    async def event_stream():
        docs = retrieve_docs(request.message)
        if not docs:
            yield f"data: {{'type': 'token', 'content': '拒绝回答'}}\n\n"
            return

        # 去重来源，避免同一篇文章引用多次
        seen = set()
        sources = []
        for doc in docs:
            key = doc.metadata.get("source", "")
            if key and key not in seen:
                seen.add(key)
                sources.append({"title": ..., "source": key})

        async for chunk in llm.astream(messages):
            yield f"data: {{'type': 'token', 'content': chunk.content}}\n\n"

        yield f"data: {{'type': 'sources', 'sources': sources}}\n\n"
        yield "data: {'type': 'done'}\n\n"

    return StreamingResponse(event_stream(), media_type="text/event-stream")