两步式 RAG：固定检索、上下文组装、生成和来源引用

问题 → Retriever → 候选过滤/去重 → 上下文组装 → LLM → 答案与引用

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableLambda


def build_context(docs, max_chars: int = 2400) -> dict:
    blocks, sources, used = [], [], 0
    for index, doc in enumerate(docs, start=1):
        source = doc.metadata.get("source", "unknown")
        text = doc.page_content.strip()
        block = f"[S{index}] source={source}\n{text}"
        if used + len(block) > max_chars:
            break
        blocks.append(block)
        sources.append({"id": f"S{index}", "source": source})
        used += len(block)
    return {"context": "\n\n".join(blocks), "sources": sources}


prompt = ChatPromptTemplate.from_messages([
    ("system", """只根据 <context> 回答。每个事实使用 [S数字] 引用。
证据不足时只回答“现有知识库无法确认”，不得使用参数知识补全。"""),
    ("human", "<context>\n{context}\n</context>\n问题：{question}"),
])


def retrieve_and_prepare(inputs: dict) -> dict:
    docs = retriever.invoke(inputs["question"])
    prepared = build_context(docs)
    return {"question": inputs["question"], **prepared}


chain = RunnableLambda(retrieve_and_prepare) | prompt | model | StrOutputParser()
print(chain.invoke({"question": "关键告警需要通知谁？"}))