文明6 RAG 知识库：架构设计与实现

from bs4 import BeautifulSoup
from langchain_core.documents import Document
import requests

base_url = 'https://www.civilopedia.net/zh-CN/gathering-storm/greatpeople/'
url = f"{base_url}great_person_individual_{person_id}"
response = requests.get(url)
soup = BeautifulSoup(response.content, 'html.parser')

# 提取关键字段
name = soup.find('div', class_='App_pageHeaderText__SsfWm').get_text()
# 提取特色能力、身份等...

doc = Document(
    page_content=f"伟人姓名:{name}\n特色能力:{abilities}\n身份:{duty}",
    metadata={"source": url}
)

from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Milvus
from langchain_ollama import OllamaEmbeddings

# 分词：chunk=1000, overlap=200
text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
split_docs = text_splitter.split_documents(docs)

# 向量化入库
embeddings = OllamaEmbeddings(model="embeddinggemma")
vectorstore = Milvus.from_documents(
    split_docs,
    embedding=embeddings,
    collection_name="great_person"
)

from pydantic import BaseModel, Field
from typing import List, Literal

class RouteQuery(BaseModel):
    tables: List[Literal[
        "belief", "religion", "country", "leader", "building", "wonder",
        "resource", "citystate", "district", "feature", "great_person",
        "improvement", "moment", "unit", "unit_promotion"
    ]] = Field(description="这个问题与哪些关键字有关，返回对应的表名")

import os
from langchain_ollama import ChatOllama, OllamaEmbeddings
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# 第一次调用：让 LLM 分析问题
llm = ChatOllama(model="gpt-oss:120b-cloud")
first_llm = ChatOllama(model="llama3.1")
prompt = ChatPromptTemplate.from_template(template)
chain = prompt | first_llm | StrOutputParser()
response = chain.invoke({"question": question})

# 第二次调用：结构化输出保证 JSON 稳定
structured_llm = ChatOllama(model="llama3.1")
structured_chain = structured_llm.with_structured_output(RouteQuery)
result = structured_chain.invoke(response)
# result.tables 即确定的检索范围

from langchain_core.runnables import RunnableParallel
from langchain_ollama import OllamaEmbeddings
from lib import get_vectorstore_from_milvus

embeddings = OllamaEmbeddings(model="embeddinggemma")

# 为每个涉及的表创建检索器
retriever_dict = {}
for table in result.tables:
    vectorstore = get_vectorstore_from_milvus(
        embeddings,
        collection_name=table,
        connection_args={"host": MILVUS_HOST, "port": MILVUS_PORT}
    )
    retriever_dict[table] = vectorstore.as_retriever(search_kwargs={"k": 100})

# RunnableParallel 并行执行，结果自动聚合
multi_retriever = RunnableParallel(retriever_dict)

from langchain_core.runnables import RunnableParallel, RunnablePassthrough
from langchain_core.prompts import ChatPromptTemplate
from langchain_ollama import ChatOllama
from langchain_core.output_parsers import StrOutputParser

def generate_answer_by_multiple_retriever(question, multiple_retriever, llm):
    template = """..."""

    prompt = ChatPromptTemplate.from_template(template)

    rag_chain = (
        RunnableParallel({
            "context": multiple_retriever,
            "question": RunnablePassthrough(),
        })
        | prompt
        | llm
        | StrOutputParser()
    )
    return rag_chain.invoke(question)

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class AskRequest(BaseModel):
    question: str

class AskResponse(BaseModel):
    answer: str

@app.post("/ask", response_model=AskResponse)
def ask(req: AskRequest):
    answer = generate_answer(req.question.strip())
    return AskResponse(answer=answer)

// POST /ask
// 请求
{ "question": "哪些领袖信仰天主教" }

// 响应
{ "answer": "..." }