实用提示：无需猜测的RAG技术：标准化的LangGraph+
《实用笔记》操作指南：无需猜测的RAG技术——为采用该模式的团队提供的标准化LangGraph+框架，包含合约、校验机制以及可直接插入的代码模块。
本指南将逐步构建从原材料到可运行系统的完整流程，适用于“无需猜测的RAG：标准化的LangGraph + LlamaIndex架构”。重点在于可操作的步骤、明确的检查点，以及可直接放入代码库的代码，无需猜测其用途。在修改代码之前，应先明确输入参数、各步骤的负责人以及完成标准。操作人员应能够从已知的检查点重新运行相应步骤，而无需猜测隐藏状态。需同时记录正常流程与异常恢复流程。重试机制、人工审核环节以及错误处理都是产品不可或缺的部分，而非后续需要补充的功能。
本文的编写目的
在撰写“为何需要这篇文章”时，首先列出相关约定：所需的输入参数、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 优先选择小型且可测试的单元，而非庞大的脚本。当某个步骤出错时，错误应指向单一责任模块，而非复杂的流程链。 在调整提示词之前，先使用固定的问题集来测试召回率。仅仅更换提示词很难解决检索效果不佳的问题。
A部分：理解LlamaIndex（先了解概念）
在完成A部分“理解LlamaIndex（先了解概念）”时，首先列出相关约定：所需输入、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 将这一阶段视为输入与验证后输出之间的契约。为相关成果命名，明确成功判定标准，杜绝无声的半完成状态。 在调整提示词之前，先使用固定的问题集来衡量召回率。仅仅更换提示词很难改善较差的检索效果。
LlamaIndex的实际功能
在研究“LlamaIndex的实际功能”时，首先需明确接口规范：所需输入、成功标志以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 在功能结果之外，还需记录处理时间以及Token或查询成本。提前了解成本情况，可避免从演示环境过渡到共享环境时出现意外费用。 在调整提示词之前，先使用固定问题集测试召回率。仅仅更换提示词很难改善较差的检索效果。 在研究“LlamaIndex的实际功能”时，首先需明确接口规范：所需输入、成功标志以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 需同时记录正常流程和异常恢复流程。重试机制、人工审核环节以及错误处理都属于产品功能的一部分，而非后续的优化工作。
五阶段RAG处理流程
将五阶段RAG流程视为可度量的体系来处理时，其效果最佳。在扩大范围之前，先记录一份理想的处理结果、一个失败案例以及回滚说明。 优先选择小型且可测试的单元，而非庞大的脚本。当某个步骤出现故障时，故障应能指向具体的责任主体，而非整个复杂的流程。 将分块策略与检索策略分开。当质量指标发生变化时，修改其中一项不应迫使重新编写另一项。
1. LOAD → Read raw files (PDF, Word, web pages, databases) into Documents
2. CHUNK → Split Documents into small, retrievable Nodes
3. EMBED → Convert each Node's text into a vector (a list of numbers
representing meaning)
4. STORE → Save those vectors in a Vector Index for fast lookup
5. RETRIEVE → At query time, embed the user's question, find the most
similar Nodes, and return them as context
你需要了解的核心关键词
《你需要掌握的核心关键词》这一方法在被视为可度量的指标时效果最佳。在扩大范围之前，先收集一份优秀的文本样本、一个失败案例以及回滚说明。 将此阶段视为输入与经过验证的输出之间的契约。为相关成果命名，明确成功标准，绝不允许出现悄无声息的半完成状态。 将分块策略与检索策略分开。当质量指标发生变化时，修改其中一项不应迫使重新编写另一项。
第二部分：构建独立的 LlamaIndex 知识库
第B部分：构建独立的LlamaIndex知识库时，若能将其视为可度量的对象，效果会更好。在扩大范围之前，需记录一份理想的处理结果、一个失败案例以及回滚说明。 在功能结果旁还需记录处理时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。 应将分块策略与检索策略分开。当质量指标发生变化时，调整其中一个不应迫使重新编写另一个。 第B部分：构建独立的LlamaIndex知识库时，若能将其视为可度量的对象，效果会更好。在扩大范围之前，需记录一份理想的处理结果、一个失败案例以及回滚说明。 需同时记录正常流程与故障恢复流程。重试机制、人工审核环节以及错误处理方式都是产品的一部分，而非后续需要补充的内容。
步骤1：安装
对于第一步：安装，应在修改代码之前明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。相比冗长的脚本，更应采用小型且可测试的单元。当某一步骤失败时，故障原因应能指向单一责任模块，而非复杂的流程链。需引用实际作为答案依据的段落；没有引用的话，操作人员就无法区分是虚假信息还是索引缺失所致。
# Core package + OpenAI LLM and embedding integrations (the common starting setup)
pip install llama-index-core llama-index-llms-openai llama-index-embeddings-openai
# Readers for common file types (PDF, Word, etc.)
pip install llama-index-readers-file pypdf
第二步：使用Settings进行全局配置
对于第二步：使用Settings进行全局配置，在修改代码之前需明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。
将此阶段视为输入与已验证输出之间的契约。为相关成果命名，定义成功判定标准，并拒绝默许的半完成状态。
需引用实际作为答案依据的段落。没有引用的话，操作人员就无法区分幻觉内容与索引缺失的情况。
# ── llamaindex_config.py ─────────────────────────────────────
import os
from llama_index.core import Settings
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding
# Settings is global - configure once, used everywhere in LlamaIndex
Settings.llm = OpenAI(
model="gpt-4o-mini", # Used for generating final answers from retrieved context
temperature=0.1, # Low temperature: factual, not creative
)
Settings.embed_model = OpenAIEmbedding(
model="text-embedding-3-small", # Used to convert text into vectors
)
# Controls how documents are split into Nodes (chunks)
Settings.chunk_size = 512 # Max tokens per chunk
Settings.chunk_overlap = 50 # Overlap between consecutive chunks, to preserve context across boundaries
第三步：加载 → 索引 → 查询（独立流程）
对于第三步：加载→索引→查询（独立处理流程），在修改代码之前需明确输入内容、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。除了功能结果外，还需记录处理时间以及令牌或查询成本。提前了解成本情况可避免在从演示环境切换到共享环境时出现意外费用。必须注明实际用于得出答案的对应内容；若没有引用，操作人员就无法区分是幻觉结果还是索引缺失导致的错误。
# ── build_knowledge_base.py ──────────────────────────────────
from llama_index.core import SimpleDirectoryReader, VectorStoreIndex
# ── LOAD: Read all files in a folder into Document objects ──
documents = SimpleDirectoryReader("./data").load_data()
print(f"Loaded {len(documents)} documents")
# ── CHUNK + EMBED + STORE: all three happen inside this one call ──
# VectorStoreIndex automatically:
# 1. Splits each Document into Nodes (using Settings.chunk_size)
# 2. Embeds each Node (using Settings.embed_model)
# 3. Stores the vectors in an in-memory index
index = VectorStoreIndex.from_documents(documents, show_progress=True)
# ── RETRIEVE + GENERATE: ask a question ──────────────────────
query_engine = index.as_query_engine(
similarity_top_k=3, # Retrieve the 3 most relevant chunks for each query
)
response = query_engine.query("What is our refund policy for enterprise customers?")
print(response)
对于第3步：加载→索引构建→查询（独立处理流程），在修改代码之前需明确输入参数、该步骤的负责人以及终止条件。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 需同时记录正常流程与异常恢复流程。重试机制、人工审核环节以及错误消息处理都是产品功能的一部分，而非后续需要补充的内容。
第4步：索引的持久化存储（无需每次都重新嵌入）
在完成第4步“持久化索引（无需每次重新嵌入）”时，首先写下相关契约：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 建议使用小型、可测试的单元而非庞大的脚本。当某一步骤失败时，故障应指向单一的责任模块，而非复杂的流程链。 在调整提示词之前，先使用固定的问题集来测试召回率。仅仅更换提示词很难改善较差的检索效果。
# ── Save the index after building it ─────────────────────────
index.storage_context.persist(persist_dir="./storage")
# ── Load it back later without re-embedding anything ──────────
from llama_index.core import StorageContext, load_index_from_storage
storage_context = StorageContext.from_defaults(persist_dir="./storage")
index = load_index_from_storage(storage_context)
第5步：使用外部向量数据库（Chroma）
在完成第5步“使用外部向量数据库（Chroma）”时，首先需列出相关约定：所需输入、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改始终符合初始要求。 将这一阶段视为输入与验证后输出之间的契约。为相关成果命名，明确成功判定标准，杜绝无声的半完成状态。 在调整提示词之前，先使用固定的问题集来测试召回率。仅仅更换提示词往往无法改善较差的检索效果。
# ── Using Chroma as a persistent, production-grade vector store ──
# pip install llama-index-vector-stores-chroma chromadb
import chromadb
from llama_index.vector_stores.chroma import ChromaVectorStore
from llama_index.core import StorageContext, VectorStoreIndex
chroma_client = chromadb.PersistentClient(path="./chroma_db")
chroma_collection = chroma_client.get_or_create_collection("my_knowledge_base")
vector_store = ChromaVectorStore(chroma_collection=chroma_collection)
storage_context = StorageContext.from_defaults(vector_store=vector_store)
# Build the index directly into Chroma
index = VectorStoreIndex.from_documents(
documents,
storage_context=storage_context
)
# Later, in a different process, reconnect without re-indexing:
index = VectorStoreIndex.from_vector_store(vector_store=vector_store)
C部分：将LlamaIndex与LangGraph连接起来
在完成C部分“将LlamaIndex与LangGraph连接”时，首先需明确接口规范：所需输入、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免从演示环境过渡到共享环境时出现意外费用。 在调整提示词之前，先使用固定问题集测试召回率。仅仅更换提示词很难改善较差的检索效果。 在完成C部分“将LlamaIndex与LangGraph连接”时，首先需明确接口规范：所需输入、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 需同时记录正常流程和异常恢复流程。重试机制、人工审核环节以及错误处理都属于产品功能的一部分，而非后续需要补充的内容。
The Bridge：将查询引擎封装为工具
“The Bridge：将查询引擎封装为工具”这一方法在被视为可度量的对象时效果最佳。在扩大范围之前，先记录一份理想的处理结果、一个故障案例以及回滚说明。 优先选择小型且可测试的单元，而非庞大的脚本。当某个步骤出现故障时，故障应指向单一责任模块，而非复杂的处理流程。 将分块策略与检索策略分开。当质量指标发生变化时，修改其中一项不应迫使重新编写另一项。
# ── MODULE 3: TOOLS (LlamaIndex-backed) ─────────────────────
from langchain_core.tools import tool
# The query_engine built in Part B - created once, at startup
# (In a real app, you'd load this from persisted storage, not rebuild it every time)
@tool
def search_knowledge_base(query: str) -> str:
"""Search the internal knowledge base for company policies, product
documentation, and internal procedures. Use this whenever the user asks
a question that might be answered by internal company documents rather
than general knowledge.
Args:
query: A natural-language question to search for.
Returns:
A synthesized answer based on the most relevant retrieved documents.
"""
response = query_engine.query(query)
return str(response)
完整集成：第1至7模块
《完整集成：第1至7模块》若作为可度量的整体来处理效果最佳。在扩大范围之前，先记录一份完美的成果文本、一个失败案例以及回滚说明。 将此阶段视为输入与已验证输出之间的契约。为相关成果命名，明确成功标准，绝不允许默默地只完成部分工作。 将分块策略与检索策略分开。当质量指标发生变化时，修改其中一项不应强制要求重新编写另一项。
# ============================================================
# LANGGRAPH + LLAMAINDEX RAG AGENT — COMPLETE TEMPLATE
# Extends: Part 1 (core structure)
# ============================================================
# ── MODULE 1: IMPORTS & CONFIGURATION ───────────────────────
import os
from typing import Literal
# LangChain / LangGraph imports (the orchestration layer)
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage, BaseMessage
from langchain_core.tools import tool
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode
from langgraph.checkpoint.memory import MemorySaver
# LlamaIndex imports (the retrieval / data layer)
from llama_index.core import (1
Settings, SimpleDirectoryReader, VectorStoreIndex,
StorageContext, load_index_from_storage
)
from llama_index.llms.openai import OpenAI as LlamaOpenAI
from llama_index.embeddings.openai import OpenAIEmbedding
# LangGraph's chat model - used by the agent's reasoning
llm = ChatOpenAI(model="gpt-4o", temperature=0)
# LlamaIndex's model config - used internally by the query engine
# Note: these are SEPARATE from the LangGraph llm above. Each framework
# manages its own model instances; they don't share state.
Settings.llm = LlamaOpenAI(model="gpt-4o-mini", temperature=0.1)
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small")
Settings.chunk_size = 512
# ── MODULE 2: STATE ──────────────────────────────────────────
class State(MessagesState):
pass # messages field inherited; extend if your agent needs more
# ── MODULE 3: TOOLS (RAG-backed) ─────────────────────────────
# Build or load the LlamaIndex knowledge base ONCE, at startup
PERSIST_DIR = "./storage"
if os.path.exists(PERSIST_DIR):
# Reload existing index - no re-embedding, fast startup
storage_context = StorageContext.from_defaults(persist_dir=PERSIST_DIR)
index = load_index_from_storage(storage_context)
else:
# First run - build the index and persist it
documents = SimpleDirectoryReader("./data").load_data()
index = VectorStoreIndex.from_documents(documents, show_progress=True)
index.storage_context.persist(persist_dir=PERSIST_DIR)
query_engine = index.as_query_engine(similarity_top_k=3)
@tool
def search_knowledge_base(query: str) -> str:
"""Search internal company documents for policies, product specs,
procedures, and other domain-specific information. Use this for any
question that requires knowledge specific to this organization rather
than general world knowledge."""
response = query_engine.query(query)
return str(response)
tools = [search_knowledge_base]
llm_with_tools = llm.bind_tools(tools)
tool_node = ToolNode(tools)
# ── MODULE 4: NODES ──────────────────────────────────────────
def agent_node(state: State) -> dict:
"""The reasoning node. Decides whether to answer directly or
search the knowledge base first."""
system_prompt = SystemMessage(content=(
"You are a helpful assistant with access to an internal knowledge base. "
"Use the search_knowledge_base tool when the user asks about company-specific "
"information. For general questions, answer directly."
))
messages = [system_prompt] + state["messages"]
response = llm_with_tools.invoke(messages)
return {"messages": [response]}
# ── MODULE 5: ROUTING ────────────────────────────────────────
def should_continue(state: State) -> Literal["tools", "__end__"]:
last_message = state["messages"][-1]
if hasattr(last_message, "tool_calls") and last_message.tool_calls:
return "tools"
return "__end__"
# ── MODULE 6: GRAPH ASSEMBLY ─────────────────────────────────
graph_builder = StateGraph(State)
graph_builder.add_node("agent", agent_node)
graph_builder.add_node("tools", tool_node)
graph_builder.add_edge(START, "agent")
graph_builder.add_conditional_edges(
"agent", should_continue,
{"tools": "tools", "__end__": END}
)
graph_builder.add_edge("tools", "agent")
graph = graph_builder.compile(checkpointer=MemorySaver())
# ── MODULE 7: ENTRYPOINT ──────────────────────────────────────
if __name__ == "__main__":
config = {"configurable": {"thread_id": "session-001"}}
print("RAG agent ready. Ask about your documents, or anything else.\n")
while True:
user_text = input("You: ").strip()
if not user_text or user_text.lower() == "exit":
break
response = graph.invoke(
{"messages": [HumanMessage(content=user_text)]},
config=config
)
print(f"Agent: {response['messages'][-1].content}\n")
运行此内容时实际会发生什么
运行此代码时实际会发生什么？若将其视为可度量的对象，效果最佳。在扩大范围之前，需记录一个理想运行案例、一个故障案例以及回滚说明。 在功能结果旁同时记录执行时间以及令牌或查询成本。提前了解成本情况，可避免从演示环境过渡到共享环境时出现意外费用。 应将分块策略与检索策略分开。当质量指标发生变化时，调整其中一个不应迫使重新编写另一个。 运行此代码时实际会发生什么？若将其视为可度量的对象，效果最佳。在扩大范围之前，需记录一个理想运行案例、一个故障案例以及回滚说明。 需同时记录正常流程与恢复流程。重试机制、人工审核环节以及死信处理都是产品的一部分，而非后续需要补充的功能。
User: "What's our policy on remote work?"
↓
[agent_node] — LangGraph's LLM reads the message, recognizes this needs
internal info, decides to call search_knowledge_base
↓
[tools] — ToolNode executes search_knowledge_base("What's our policy on remote work?")
↓
Inside the tool: query_engine.query(...) runs —
this is 100% LlamaIndex, invisible to LangGraph:
1. Embeds the query
2. Searches the vector index for the 3 closest chunks
3. Feeds those chunks + the question to Settings.llm
4. Returns a synthesized answer string
↓
[agent_node] — LangGraph's LLM receives the tool's string result,
and crafts the final response shown to the user
↓
Response to user
D部分：更深入一层——仅检索器模式（更高控制度）
在采用D部分：更深入一层——仅检索器模式（更高控制度）时，应在修改代码之前明确输入内容、该步骤的负责人以及退出标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 建议使用小型、可测试的单元而非庞大的脚本。当某个步骤失败时，故障应指向单一责任点，而非复杂的流程链。 需引用实际作为答案依据的段落。没有引用的话，操作人员就无法区分幻觉与索引缺失。
# ── Retriever-only tool: returns raw chunks, not a synthesized answer ──
retriever = index.as_retriever(similarity_top_k=3)
@tool
def retrieve_documents(query: str) -> str:
"""Retrieve relevant document excerpts from the internal knowledge base.
Returns raw excerpts for you to read and reason over yourself -
use this when you need to cite specific sources or combine information
from multiple documents."""
nodes = retriever.retrieve(query)
# Format each retrieved chunk with its source for transparency
formatted_chunks = []
for i, node in enumerate(nodes):
source = node.metadata.get("file_name", "unknown source")
formatted_chunks.append(f"[Excerpt {i+1} from {source}]\n{node.text}")
return "\n\n---\n\n".join(formatted_chunks)
何时使用QueryEngine与Retriever
在决定何时使用QueryEngine而非Retriever时，应在修改代码之前明确输入内容、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。
将此阶段视为输入与经过验证的输出之间的契约。为相关成果命名，定义成功判定标准，并拒绝默许的半完成状态。
需引用实际作为答案依据的段落。没有引用的话，操作人员就无法区分幻觉内容与索引缺失的情况。
更新后的关键词参考卡
对于更新后的关键词参考卡，在修改代码之前需明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 在功能结果旁记录执行时间以及令牌或查询成本。提前显示成本可避免在流程从演示环境转向共享环境时出现意外费用。 需注明实际作为答案依据的段落。没有引用的话，操作人员就无法区分是幻觉内容还是索引缺失导致的错误。 对于更新后的关键词参考卡，在修改代码之前需明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 需同时记录正常流程和异常恢复流程。重试机制、人工审核环节以及错误处理方式都是产品本身的组成部分，而非后续才添加的完善措施。
决策指南：何时真正需要它？
在阅读《决策指南：何时真正需要它？》时，首先写下合同条款：所需输入、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持透明。 建议使用小型、可测试的单元，而非庞大的脚本。当某个步骤失败时，故障应指向单一责任点，而非复杂的流程链。 在调整提示词之前，先在固定的问题集上测试召回率。频繁更换提示词往往无法解决检索效果不佳的问题。
结论：两种框架，无缝衔接
在撰写“结论：两种框架，一个无缝集成”时，首先明确合同条款：所需输入、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改始终符合要求。 将此阶段视为输入与验证后输出之间的契约。为相关成果命名，定义成功判定标准，杜绝无声的半完成状态。 在调整提示词之前，先使用固定的问题集来衡量召回率。仅仅更换提示词往往无法解决检索效果不佳的问题。
操作检查清单
若将操作检查清单视为可衡量的指标，其效果会更好。在扩大范围之前，先记录一份最佳示例、一个失败案例以及回滚说明。 将配置信息置于应用程序代码之外。环境文件、密钥存储和功能开关应集中存放于一个位置，以便操作人员无需查看整个系统结构即可进行审计。
应将分块策略与检索策略分开。当质量指标发生变化时，修改其中一项不应强制重新编写另一项。
对于涉及资金支出或更改生产数据的操作，必须经过人工审批。编译时的配置并不等同于业务上的完整性。
编写简短的操作手册：说明如何轮换密钥、如何清空队列以及如何回滚上一次的导入操作。
将此阶段视为输入数据与经过验证的输出结果之间的契约。为相关产物命名，明确成功标准，杜绝默许的半完成状态。
在升级整个系统之前，先冻结版本，为关键流程保存标准模板，并确认回滚步骤。共享环境需要设置速率限制、进行租户检查，同时明确密钥轮换的负责人。与其追求花哨的一次性演示，不如注重扎实的可靠性。
bcaf14f9c811的批次处理说明：不要将提供者密钥放入代码仓库，为每个会话设置令牌上限，并将转录内容存储在评估测试用例的旁边，以便后续更换模型时仍能保持可比性。