LlamaIndex Plus LangGraph：RAG智能体的简洁架构方案
如何使用 LlamaIndex 构建知识库，并将其作为工具提供给 LangGraph 智能体使用，同时介绍持久化存储、Chroma 以及仅检索模式的相关内容。
来源: RAG Without the Guesswork: A Standardized LangGraph + LlamaIndex Pattern.
盲点与分工
LangGraph智能体仅知晓其在训练过程中从大型语言模型那里学到的知识。对于内部定价政策、上季度的财务报告或上周完成的手册等问题，它要么承认不知情，要么更糟糕的是给出自以为正确的编造答案。
检索增强生成（RAG）弥补了这一缺陷。在处理查询时，它会从您的文档中检索相关片段，并在大型语言模型作答之前将这些片段作为上下文传递给模型，从而使模型基于真实信息进行回答而非猜测。RAG是添加到单个智能体上的功能，它并非新的编排模式，因此监督节点、子图以及Send功能均不参与其中。只需了解LangGraph智能体的结构以及如何为其提供工具即可。
在 LangGraph 中手动构建数据处理流程是可行的：可以自行编写分块代码、嵌入计算代码、向量索引及检索代码。许多团队都在这么做，但对于一个早已有成熟解决方案的问题而言，这实在是一项繁琐的工作。LlamaIndex 的出现就是为了承担这些工作。在处理 AI 应用程序的数据相关任务——包括数据加载、分块、嵌入、索引和检索——方面，它无疑是最为成熟且流行的选择。而涉及状态管理、控制流以及多步骤推理的编排工作则由 LangGraph 负责。
正是基于这种分工，大多数生产环境中的 RAG 系统都采用了这样的架构：LlamaIndex 负责创建并提供知识库，而 LangGraph 则将其视为一种工具，并决定何时调用它。
LlamaIndex 的思维模型
可以把 LlamaIndex 看作是你的大语言模型的图书管理员。大语言模型就像一位出色的研究助手，它阅读过大量互联网内容，但从未接触过你公司的文件，而且也没人能要求它记住一本一万页厚的资料册。于是图书管理员便介入其中：它读取每份文档，将其拆分成便于搜索的小索引卡片（数据块），把这些卡片存储在能快速检索主题的地点（向量索引），每当助手提出问题时，它就只调取3到5张最相关的卡片。助手永远不会记住任何内容，只有少量片段会被提供，推理就在这些片段之上进行。
五阶段处理流程
无论采用何种框架，所有的 RAG 系统都会经历相同的五个阶段，而 LlamaIndex 为每个阶段都提供了简洁的抽象实现。
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
前四个阶段在构建知识库时执行一次，这一过程被称为索引构建。第五个阶段则在用户提出查询时启动，称为查询。在RAG系统中，索引构建时间与查询时间的关系可能是最重要的概念模型，值得牢记。
词汇表
文档指的是原始的内容片段，如一份PDF、一个网页或一条数据库记录，以及其元数据。它是未经处理、完整的单元。节点是从文档中截取出的片段。大语言模型和嵌入模型处理简短、聚焦的文本效果更好，因此LlamaIndex会将每个文档拆分成多个节点对象。节点会将元数据传递回其所属的父节点，从而可以追踪任何片段的来源。
SimpleDirectoryReader是常用的加载器：只需将其指向某个文件夹，它就能读取PDF、文本、Word文件以及其他格式的文件，每份文件会生成一个或多个Document对象。VectorStoreIndex是主要的索引类。给定Document或Node对象后，它会将这些对象嵌入其中，并在内存中或外部向量数据库中构建可搜索的向量索引。Settings是一个全局配置对象。大语言模型和嵌入模型只需在其中设置一次，所有组件都会自动获取这些配置，因此无需在每次调用时重新传入。QueryEngine 是实际用于提问的对象。它来自索引的 .as_query_engine() 方法，一次调用即可嵌入查询、执行相似度搜索，并将检索到的上下文传递给大语言模型以生成最终答案。Retriever 来自索引的 .as_retriever() 方法，仅负责搜索步骤。它会返回原始的匹配 Node 对象，不会生成答案，这适用于需要将检索与生成视为独立、可控制步骤的场景。而 LangGraph 正是用于处理这类情况的工具。该流程首先会在不使用 LangGraph 的情况下构建起来。
仅需约十行代码即可实现的完整流程
首先需要进行安装。
# 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
LlamaIndex 故意将功能拆分为多个小型包。llama-index-core 包含框架逻辑，而不同的大语言模型、嵌入模型、向量数据库以及文件读取器的选择则分别位于独立的包中，仅在需要时才进行安装。这样能保持安装包的轻量级。
一个简单的配置文件用于设置全局默认值。
# ── 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
关于chunk_size：过大的数据块包含过多文本，导致检索结果不准确，因为取出的某个数据块可能只有10%的相关性。过小的数据块则会丢失大语言模型理解事实所需的上下文。通常以512个标记为起点，并设置50个标记的重叠量，再根据具体文档情况进行调整。
接下来是知识库本身：
# ── 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)
大约十行代码就能构建出一个可用的RAG系统，这既是该框架的“Hello World”示例，也充分说明了它存在的意义。若要手动实现，则需要自定义分词器、用于调用嵌入模型的循环、最近邻搜索功能以及提示词模板。
索引的持久化
在构建索引时，每个数据块都需要经过嵌入模型处理，这既耗费资金又浪费时间，因此每次应用启动时都重新构建索引是极其浪费的。将索引保存到磁盘并在之后加载即可避免这一问题。
# ── 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)
storage_context.persist()会将向量数据、节点元数据以及文档存储到本地文件夹中，而load_index_from_storage()则无需重新嵌入数据或调用API，就能从这些文件中重建出完全相同的索引。它的设计理念与LangGraph的checkpointer相同，即只需构建一次，之后即可低成本地持久化存储和重新加载。
真正的向量数据库
对于原型开发来说，内存中的索引已经足够。但在实际生产环境中，通常需要专为处理数百万条向量、高并发读取以及大规模元数据过滤而设计的专用向量数据库。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)
这种切换方式类似于检查点机制。只需更换一个组件（即ChromaVectorStore的内存存储方式，就像MemorySaver被SqliteSaver取代一样），而后端的查询引擎和检索逻辑则保持不变。Pinecone、Qdrant、Weaviate以及LlamaIndex支持的大多数其他向量数据库都是以相同方式运行的。
连接两种框架
问题的关键在于：这两种框架使用不同的语言——LlamaIndex的QueryEngine并非LangGraph的工具，因为LangGraph的智能体要求使用LangChain的@tool装饰器构建的工具。
这个桥接方式很简单。定义一个普通的 Python 函数，在其上加上 @tool 注解，然后让该函数的主体调用 query_engine.query(...) 即可。这就是全部的集成步骤。LangGraph 无需了解 LlamaIndex 的内部实现，因为它看到的只是一个接收字符串并返回字符串的工具，与其他工具并无不同。
# ── 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)
文档字符串对于任何工具来说都同样重要。智能体的大语言模型会阅读它，以决定何时调用此工具，而不是依靠自身的知识回答或调用其他工具。
完整的智能体
这是完整的智能体模板，其中在工具模块中加入了由LlamaIndex支持的RAG工具，其他部分均保持不变。
# ============================================================
# 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
其架构十分简洁。LangGraph 永远不会遇到 Node、嵌入向量或数值向量，只有能够返回文本的工具，这与 search_web 或 calculate 没有区别。LlamaIndex 也不存在 messages、state 或 thread_id 这类概念，仅有输入的查询字符串和输出的答案。每个框架都发挥自身的优势，而一次函数调用则构成了两者之间的衔接点。
仅使用检索器的模式以实现更多控制
query_engine.query() 这种模式是最简单的集成方式，适用于大多数场景。但有时更需要更多控制，比如让 LangGraph 智能体的自有大语言模型利用现有的系统提示和对话上下文来撰写最终答案，而非依赖 LlamaIndex 内部的 Settings.llm，因为后者会生成独立且不连贯的答案。
在这种情况下，可用 Retriever 代替 QueryEngine。它仅执行搜索步骤并返回原始数据块，后续的处理则由智能体负责。
# ── 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)
通过这种设置，agent_node会使用llm_with_tools结合完整的对话历史和系统提示来调用自身的LLM，读取原始摘录内容并生成答案。这样做有实际好处：智能体能够明确指出某条事实出自哪份文档，还能将检索到的摘录与之前的对话内容混合，这样用户听到的就是统一的声音，而非两种风格各异的LLM输出。
该如何选择
如果只需LlamaIndex的LLM以独立字符串的形式生成基于检索的答案，且集成较为简单，可选择QueryEngine。这适用于大多数聊天机器人和助手。
当需要引用来源信息、智能体的主大语言模型应在检索到的答案与未检索到的答案之间保持一致的语气和推理风格，或是在生成最终答案之前必须将检索到的内容与其他工具的输出结合时，请选择Retriever。
关键词参考
加载与索引。 Document 是已加载的源文件，即一个文件或一个页面。Node 是其组成部分，是可检索的单位，带有指向其来源的元数据。SimpleDirectoryReader("./path").load_data() 会将该文件夹中所有支持的文件作为 Document 对象加载。Settings 是全局配置的单例，Settings.llm、Settings.embed_model 和 Settings.chunk_size 的值会为所有组件设置一次。VectorStoreIndex.from_documents(documents) 可以一次性完成分块、嵌入和存储操作。
查询功能。 index.as_query_engine(similarity_top_k=3) 会生成一个 QueryEngine，它不仅能获取排名前 k 的内容块，还会生成综合答案。index.as_retriever(similarity_top_k=3) 会生成一个 Retriever，仅用于获取排名前 k 的内容块，这种设计适用于由自定义大语言模型负责内容生成的场景。query_engine.query("...") 会执行完整的“检索后再生成”流程，并返回一个 Response 对象，而 str(response) 可用于提取其中的文本。retriever.retrieve("...") 仅执行检索操作，返回的 Node 对象包含 .text 和 .metadata 属性。
持久性。 index.storage_context.persist(persist_dir="./storage") 会保存索引，这样重启时就不必重新嵌入。通过 StorageContext.from_defaults(persist_dir="./storage") 结合 load_index_from_storage(...) 即可重新加载该索引。ChromaVectorStore、PineconeVectorStore 和 QdrantVectorStore 是外部后端，用于替代默认的内存存储，其作用与 LangGraph 的检查点机制类似。
桥梁作用。一个其函数体调用query_engine.query(...)的@tool函数构成了整个集成方案：LangGraph将其视为普通工具，而LlamaIndex则保持封装状态。此外还存在两种独立的LLM配置。Settings.llm是LlamaIndex内部的模型，仅在查询引擎中使用，而代理的LangGraph llm则负责核心推理工作。二者独立运行，不会共享任何历史记录或状态。
你需要这个吗？
- 如果代理仅需通用知识、数学能力或实时网络搜索功能，就不需要LlamaIndex，使用带有标准工具的基本模板即可。
- 如果它必须从诸如政策文件、手册或极少更新的报告等固定文档集中获取答案，将
QueryEngine封装为工具正是合适的选择。
Retriever 模式。interrupt() 函数，将检索调用置于等待审批的节点中，就像使用其他敏感工具一样。两种框架与一个衔接点
其核心启示在于：为智能体赋予现实世界知识并非 LangGraph 的问题，而是一个数据问题，数据框架才是解决它的合适工具。复杂的多智能体系统实际上是由多个单智能体图组合而成的，知识获取同样遵循这种模块化思路。
LlamaIndex 负责索引与检索，LangGraph 负责推理与协调。二者之间的连接通过一个带有 @tool 注解的函数实现。这并非限制，而是恰到好处的耦合程度——因为每个框架都能各司其职，且双方均可独立升级。向量数据库可以更换而不影响图结构，也可以添加监督模式而不改动检索逻辑。
相关阅读
- LangGraph Agent 的 MCP：从基础到实战——服务器、传输机制与适配器桥梁 — 从主机、客户端和服务器的角色出发，学习模型上下文协议，了解 FastMCP 服务器、stdio 与 HTTP 的区别、langchain-mcp-adapters 的用法，以及 LangGraph 在实际应用中可能遇到的问题。
- 构建 ShopBot：具备记忆功能、工具支持及退款审批功能的 LangGraph 客服智能体 — 通过模块化方式在 LangGraph 中构建客服智能体，涵盖状态管理、工具使用、节点构建、路由逻辑、滚动摘要功能，以及用于等待退款审批的中断机制。