FastAPI + LlamaIndex + DeepSeek 实战：RAG 文档检索与多工具 Function Calling 全拆解

FastAPI + LlamaIndex + DeepSeek 实战：RAG 文档检索与多工具 Function Calling 全拆解

前言介绍

本文把项目里两段真实代码拆开讲：

• 第一段：基于 LlamaIndex 的 RAG 管道——加载 Word 和 PDF 文档、按 token 级分片、转嵌入向量存入 ChromaDB，再加载索引用 DeepSeek 做问答，检索结果带来源和相似度分数；
• 第二段：FastAPI 路由里的三工具 Function Calling——学历查询、岗位查询、知识库检索（复用已入库的 ChromaDB），模型自动选工具、执行并发回灌，路由对外暴露接口供客户端调用。

功能概览

一、环境准备：依赖下载与配置（单列）

环境准备涉及两个库组，配置单独拎出来讲。

1.1 依赖下载

# LlamaIndex RAG 管道
pip install llama-index llama-index-vector-stores-chroma llama-index-embeddings-openai
pip install llama-index-llms-openai llama-index-llms-deepseek chromadb

# FastAPI 工具路由
pip install openai redis chromadb python-docx fastapi

• llama-index 系列：LlamaIndex 核心 + Chroma 向量存储 + 嵌入 + LLM 适配器；
• chromadb：向量库，两端代码都用 PersistentClient 落本地磁盘；
• openai：OpenAI 官方 SDK，工具路由直接用它调百炼；
• redis：学历查询缓存 + 多轮对话（db=10）；
• python-docx：只被最早的知识库入库脚本用到，LlamaIndex 本身不用。

1.2 密钥配置（环境变量）

raw_key = os.getenv("DASHSCOPE_API_KEY")
api_key = raw_key.strip()

百炼 API Key 读自 DASHSCOPE_API_KEY 环境变量。LlamaIndex 索引构建脚本里多了判空保护（if not … raise ValueError），工具路由里直接 .strip() 没判空——取不到密钥时后者直接 None.strip() 崩掉，是配置阶段最常见的事故。

1.3 base_url 兼容端点

BASE_URL = "https://ws-xfbad2xxmhwytrdx.cn-beijing.maas.aliyuncs.com/compatible-mode/v1"

这是北京地域 MaaS 工作空间专属地址，不是通用的 dashscope.aliyuncs.com。LlamaIndex 里通过 api_base 参数传进去，OpenAI SDK 里通过 base_url。地域配错、或漏写 /compatible-mode/v1 后缀，都会直接连不上，报错往往很抽象。

二、知识点讲解

2.1 为什么用 LlamaIndex 而不是自己手写 RAG

之前项目里手写过一遍 RAG：python-docx 读 Word、自己按字符切片、调嵌入 API、手动拼 ChromaDB。LlamaIndex 把这些步骤全抽象成了配置项——换向量模型改一行 Settings.embed_model，换 LLM 改一行 Settings.llm，文档格式从 docx 到 PDF 不用写额外解析代码。代价是不理解内部机制时出了错很难排查。

2.2 SentenceSplitter 按 token 分片 vs 按字符分片

之前手写的 split_text 是按 200 字符硬切，中文一句话可能在 200 字符处被拦腰截断。LlamaIndex 的 SentenceSplitter 按 token 数分片，还带 chunk_overlap（重叠量），确保关键句子不会因为被切开而丢失语义。这是 RAG 检索效果的一个分水岭。

2.3 Function Calling 的系统提示词

给模型一个 system 消息，在里面写明"什么时候该调哪个工具"，模型在判断是否需要工具时会更准确。工具路由开头就有一段 system_prompt，明确告诉模型三种工具各自什么场景下用。

2.4 同步/异步工具混用与 hasattr(…, "__await__")

工具路由里三个工具：两个同步（学历查询、知识库查询）、一个异步（岗位查询查 MySQL）。收结果时用 hasattr(tool_result, "__await__") 判断是否是协程，是就 await，不是就直接用——这样同步工具和异步工具可以放在同一个 func_mapping 字典里统一调用。

三、代码逻辑拆解

第一部分：LlamaIndex RAG 管道

3.1 全局配置（向量模型 + LLM + 文本分割器）

索引构建脚本的配置段：

Settings.embed_model = OpenAIEmbedding(
mode="similarity",
model="text-embedding-ada-002",
model_name="qwen3.7-text-embedding",
api_key=DASHSCOPE_API_KEY,
embed_batch_size=20,
api_base=BASE_URL,
max_retries=3,
timeout=60,
)

Settings.llm = OpenAI(
model="qwen-plus",
api_key=DASHSCOPE_API_KEY,
api_base=BASE_URL,
max_retries=3,
timeout=60,
)

• Settings.embed_model：LlamaIndex 全局向量模型。model="text-embedding-ada-002" 是 LlamaIndex 内部的模型类标识，实际调的百炼模型是 model_name="qwen3.7-text-embedding"。如果把 model 写成别的类，LlamaIndex 可能走不同的参数拼装逻辑，但最终 API 发出的还是 model_name 这个值；
• embed_batch_size=20：嵌入时一次发 20 条，和之前手写 RAG 里 batch_size = 20 逻辑一致；
• Settings.llm：索引构建阶段大模型用 qwen-plus，注意它只在建索引时用到（做文档摘要等），查询阶段会另外指定；
• S.max_retries=3, timeout=60：重试和超时，LlamaIndex 默认不设这些，百炼偶发超时时没配就会直接抛异常。

文本分割器配置：

Settings.text_splitter = SentenceSplitter(
chunk_size=500,
chunk_overlap=150,
)

• chunk_size=500：每片约 500 个 token，按 token 不是按字符——块太大检索不准、太小上下文不足；
• chunk_overlap=150：相邻块之间有约 150 token 重叠。比如文档切出来是 [0-500] [350-850] [700-1200…]，重叠段确保"转正标准是连续三个月绩效达标"这种跨块句子不会被切断。这里容易踩的坑是：重叠量设太小（如 0 或 50），检索时会发现明明答案在文档里，但抓出来的片段恰好少了关键半句。

3.2 文档加载

def load_documents(data_dir="./data"):
reader = SimpleDirectoryReader(
input_dir=data_dir,
recursive=True,
file_metadata=lambda filename: {"source": os.path.basename(filename)},
)
documents = reader.load_data()
print(f"加载了 {len(documents)} 个文档片段")
return documents

• SimpleDirectoryReader：LlamaIndex 的"万能读文档"——把目标目录下所有文件（.docx、.pdf、.txt 等）自动识别格式并切成文档对象。项目 data 目录里有 employee_manual.docx 和一份浦发银行 PDF，两种格式一次加载，不用写任何区分代码。
• recursive=True：递归读取子目录；
• file_metadata lambda：给每个文档对象的元数据打上来源文件名，查询时能看到答案来自哪份文件（node.metadata.get('source')）。如果不上这个溯源标记，后面查询时打印来源是"未知"，排查答案准确性时就差一步信息。

3.3 索引构建与持久化

def build_and_save_index(documents, persist_dir="./storage"):
chroma_client = chromadb.PersistentClient(path=persist_dir)
chroma_collection = chroma_client.get_or_create_collection(
name="bank_policy_collection"
)
vector_store = ChromaVectorStore(chroma_collection=chroma_collection)
storage_context = StorageContext.from_defaults(vector_store=vector_store)

index = VectorStoreIndex.from_documents(
documents,
storage_context=storage_context,
show_progress=True,
)

index.storage_context.persist(persist_dir=persist_dir)
vector_count = chroma_collection.count()
print(f"索引已保存到 {persist_dir}，向量数: {vector_count}")
return index

• chromadb.PersistentClient(path=persist_dir)：ChromaDB 数据直接落进 persist_dir 目录（这里是 ./storage）；
• get_or_create_collection(name="bank_policy_collection")：集合名称是 bank_policy_collection（银行制度），与之前员工手册的知识库用不同集合名区分；
• ChromaVectorStore + StorageContext.from_defaults：把 Chroma 向量存储接入 LlamaIndex 的存储上下文——这里不传 persist_dir 给 StorageContext（注释写明了），避免启动时加载旧数据冲突；
• VectorStoreIndex.from_documents：批式把文档嵌入并入库，show_progress=True 显示进度条；
• index.storage_context.persist(persist_dir=persist_dir)：除了 ChromaDB 里的向量，LlamaIndex 还会落 docstore.json、index_store.json、graph_store.json 等元数据——这些文件是查询阶段 load_index_from_storage 能正确还原索引的前提，缺了任意一个都会加载失败；
• 最后打了一个向量数量验证日志，可以快速确认入库是否完整。

3.4 索引加载（查询脚本的配置差异）

查询脚本的向量模型配置与建索引脚本完全一致——这是必须的，因为查询时要把用户问题转成向量，嵌入模型必须和入库时用的是同一个，否则向量不在同一语义空间、检索结果毫不相关。但 LLM 换成了 DeepSeek：

from llama_index.llms.deepseek import DeepSeek

Settings.llm = DeepSeek(
model="deepseek-v4-flash",
api_key="sk-xxx", # 实际代码里为真实密钥，此处打码
)

• 查询阶段用 deepseek-v4-flash 做问答生成（注意不是 deepseek-v4-pro 或 deepseek-v4-flash-0731）；
• API Key 是硬编码在代码里的——sk-7a28ce25…，这是个真实密钥。提交到仓库即泄露、换环境还要改代码，应改为 os.getenv("DEEPSEEK_API_KEY")。

索引加载逻辑：

def load_index(persist_dir="./storage"):
chroma_client = chromadb.PersistentClient(path=persist_dir)
chroma_collection = chroma_client.get_collection("bank_policy_collection")
vector_store = ChromaVectorStore(chroma_collection=chroma_collection)
storage_context = StorageContext.from_defaults(
vector_store=vector_store,
persist_dir=persist_dir
)
index = load_index_from_storage(storage_context)
print(f"索引加载成功（向量数: {chroma_collection.count()}）")
return index

• get_collection 取同名集合（不是 get_or_create_collection）；
• StorageContext 建索引时没传 persist_dir，这里加载时必须传 persist_dir，load_index_from_storage 要靠它找到 docstore.json 等元数据文件把索引还原成可查询状态；
• load_index_from_storage 返回的 index 对象可直接调用 as_query_engine() 进行问答。

3.5 查询与检索溯源

def query_policy(index, user_query, top_k=3):
query_engine = index.as_query_engine(similarity_top_k=top_k)
response = query_engine.query(user_query)

print("\\n检索到的相关片段（RAG 依据）:")
for i, node in enumerate(response.source_nodes, 1):
print(f" [{i}] 相似度: {node.score:.4f} | 来源: {node.metadata.get('source', '未知')}")
preview = node.text.replace('\\n', ' ')[:100] + "…"
print(f" 预览: {preview}")

return response.response

• index.as_query_engine(similarity_top_k=top_k)：创建查询引擎，top_k=3 取最相似 3 条。top_k 的值直接决定答案质量——设太小（如 1）模型只看到一条片段容易信息不完整，设太大（如 10）提示词太长可能稀释关键信息或超 token 限制；
• response.source_nodes：LlamaIndex 查询结果自带检索来源，每个 node 有 score（相似度分数）、metadata.source（来自 file_metadata lambda 打的文件名）、文本内容。这段打印就是你排查"答案不准"时最重要的调试依据——能直接看到模型到底拿到了哪个片段的哪些内容。

第二部分：FastAPI 三工具 Function Calling

3.6 客户端初始化与 Redis

r = Redis(host='127.0.0.1', port=6379, db=10, decode_responses=True)

raw_key = os.getenv("DASHSCOPE_API_KEY")
api_key = raw_key.strip()
client = OpenAI(
api_key=api_key,
base_url="https://ws-xfbad2xxmhwytrdx.cn-beijing.maas.aliyuncs.com/compatible-mode/v1",
)

• Redis db=10：与多轮对话共用同一个库号，学历查询的 key 前缀 boos:llm:academic_credential_verification: 做了命名空间隔离，暂时不冲突；
• OpenAI 客户端是模块级单例——client 在模块顶层创建一次，后续所有函数复用。这和之前手写 RAG 里每次新建客户端的做法不同，避免了重复握手。

3.7 三工具定义

工具一：学历查询（同步 + Redis 缓存）

def academic_credential_verification(arguments):
if arguments is None:
return "请提供正确的学历验证码"
vcode = arguments["vcode"]
if vcode is None:
return "请提供正确的学历验证码"
key = f"boos:llm:academic_credential_verification:{vcode}"
redis_vreif = r.get(key)
if redis_vreif is None:
API_KEY = "自己的apikey"
BASE_URL = "https://www.apimy.cn/api/xxw/bgcx"
params = {"key": API_KEY, "vcode": arguments["vcode"]}
headers = {"Content-Type": "application/json"}
response = requests.get(BASE_URL, params=params, timeout=30)
response.raise_for_status()
data = response.json()
r.set(key, json.dumps(data, ensure_ascii=False))
return json.dumps(data, ensure_ascii=False)
else:
return redis_vreif

• 前 4 行做了两次判空保护（arguments is None + vcode is None），模型偶发传空参时不至于崩；
• key 用验证码拼 Redis key，同一验证码第二次查直接走缓存——这里 r.set 没设 ex（过期时间），学历数据会永久驻留 db=10。如果第三方 API 数据更新了（比如学历状态变了），缓存不失效就拿脏数据；
• API_KEY = "MY_KEY_le0KRXAsCh6cNphDEURqJCs02jt3x1" 又是硬编码在源码里的 API Key；
• requests.get(…, timeout=30)：同步阻塞 IO，FastAPI 的 async def 路由里调它会阻塞事件循环。生产环境必须换 httpx.AsyncClient，否则并发上来所有请求排队等这个同步调用。

工具二：岗位查询（异步查询 MySQL）

async def query_job_tool(argumenis):
if argumenis is None:
return "请提供正确的岗位信息参数"
job_id = argumenis["job_id"]
job = await Job.filter(id=job_id).first()
if job is None:
return "没有此岗位"
else:
job_info = f"""
职位名称:{job.job_name}
"""
return job_info

• 这是一个 async def 函数，走 Tortoise-ORM 查 MySQL 的 Job 表；
• 参数名 argumenis 是个笔误（应为 arguments），但因为第 2 行就直接用了 argumenis，功能一致就没人管了；
• 返回的 job_info 只拼了职位名称一条字段，真实场景应该把薪资、地点、要求等也拼进去，否则模型知道岗位名也不知道别的信息；
• 关键点：这个函数是异步的，和下面两个同步工具放在同一个 func_mapping 调用时需要区别处理（见 3.9）。

工具三：知识库查询（同步 ChromaDB）

def query_knowledge_tool(arguments):
user_query = arguments["user_query"]
user_query_embeddings = text_embedding(user_query)
client = chromadb.PersistentClient(path=r"E:\\shijian\\P4\\zuoye\\Boos0721\\zhishi")
collection = client.get_collection(
name="employee_manual_collection",
embedding_function=None
)
results = collection.query(
query_embeddings=user_query_embeddings,
n_results=5,
include=["metadatas", "documents", "distances"]
)
retrieval_chunks = results["documents"][0]
content = "\\n".join([f"–: {chunk}" for chunk in retrieval_chunks])
prompt = f"""
## 角色设定
你是一个专业的人力资源专家,
## 任务描述
根据用户的问题和公司的制度内容,回答问题
## 输入数据
用户的问题:{user_query}
公司制度内容:{content}
## 约束
1:严格基于公司的制度内容回答问题,不要胡编乱造
2:如果公司的制度内容没有相关信息,请明确说明"暂无此知识"
3:回答简洁明了,条理清晰
"""
client = OpenAI(
api_key=api_key,
base_url="https://ws-xfbad2xxmhwytrdx.cn-beijing.maas.aliyuncs.com/compatible-mode/v1",
)
completion = client.chat.completions.create(
model="deepseek-v4-pro",
messages=[{"role": "user", "content": prompt}],
)
return completion.choices[0].message.content

• text_embedding(user_query)：从项目里的嵌入函数直接复用——问题向量化必须和入库时用的是同一个嵌入模型（qwen3.7-text-embedding），否则向量不在同一个语义空间，检索完全对不上；
• PersistentClient(path=r"E:\\shijian\\…")：ChromaDB 路径是本地绝对路径，换机器部署直接炸。应改成相对路径或从配置读；
• collection.get_collection 取 employee_manual_collection 集合——这是最早员工手册入库时建的集合名，和前面 LlamaIndex 建的 bank_policy_collection 是两个独立的库；
• 检索 top5 片段，拼成提示词再用 deepseek-v4-pro 生成回答。这里又在函数内部新建了一个 OpenAI 客户端（有模块级单例了还要新建），重复创建连接。

3.8 系统提示词与工具映射

system_prompt = """
你是一个专业的智能助手，规则如下：
1. 当查询学历或者验证学历时，必须调用 academic_credential_verification 工具完成;
2. 当你想查询岗位相关信息时，必须调用 query_job_tool 工具完成;
2. 当你想查询企业员工手册内容时，必须调用 query_knowledge_tool 工具完成;
"""

messages = [
{"role": "system", "content": system_prompt}
]

• system_prompt 用"必须调用 xxx 工具"来约束模型行为——这是在 tools 声明之外的另一层引导，让 model 在判断工具调用时更果断；
• 注意第二条的序号打成了 2（应该是 3），这是真实代码里的小笔误，不影响功能但值得修；
• messages 是模块级全局列表，多请求并发时会串话——不同的用户请求会 append 到同一个列表里，A 能看到 B 的消息。生产环境应把 messages 放进请求作用域。

3.9 llm_chat 路由：三工具完整闭环

@llm_day05_router.post("/llm_chat")
async def llm_chat(request: LLMCase3):
user_message = {"role": "user", "content": request.user_query}
messages.append(user_message)
completion = get_ai_response(messages)
messages.append(completion.choices[0].message)
if completion.choices[0].message.tool_calls is None:
return {
"code": "1",
"msg": "直接回答",
"data": completion.choices[0].message.content,
}
else:
for i in completion.choices[0].message.tool_calls:
tool_id = i.id
func_name = i.function.name
func_arguments = i.function.arguments

func_mappinf = {
"academic_credential_verification": academic_credential_verification,
"query_job_tool": query_job_tool,
"query_knowledge_tool": query_knowledge_tool,
}
tool_result = func_mappinf[func_name](json.loads(func_arguments))
if hasattr(tool_result, "__await__"):
tool_result = await tool_result
tool_message = {
"role": "tool",
"content": tool_result,
"tool_call_id": tool_id
}
messages.append(tool_message)
completion = get_ai_response(messages)
return {
"code": "1",
"msg": "调用工具",
"data": completion.choices[0].message.content,
}

• 第 2–3 行：用户消息追加进全局 messages，发起初次请求；
• 第 4 行：get_ai_response 里模型是 qwen3.8-max——注意不是 deepseek-v4-pro。项目里不同接口用了不同模型，跨接口混用模型会导致回答风格甚至能力完全不同；
• 第 5–5 行：模型回答也 append 进 messages，保持多轮上下文；
• 第 6–11 行：tool_calls is None 说明模型没触发工具，直接返回；
• 第 12–30 行：进入工具分支：
  • 第 23 行 func_mappinf[func_name](json.loads(func_arguments))：三个工具统一从字典里按模型返回的函数名分发调用。query_job_tool 是 async def，这里直接调用会返回一个协程对象而不是结果——所以下面必须处理。
  • 第 24–25 行 hasattr(tool_result, "__await__")：判断返回值是不是协程（await 对象），是就 await 拿结果。这是让同步工具和异步工具共存在同一个映射表里的常用写法——但调用顺序是逐个的（不是 asyncio.gather），遇到多个并行 tool_calls 时异步工具不能并发执行；
  • 第 26–30 行：拼 role=tool 消息，tool_call_id 用模型给的 i.id，如果丢了 tool_call_id 模型会报"找不到对应工具调用"；
• 第 31–36 行：所有工具结果都追加完后，统一调第二次 get_ai_response(messages)，生成自然语言答案。这里是循环外统一请求，和之前学历查询的"在循环内逐个回灌"不同，避免了多并行 tool_calls 时的重复请求。