LLM + Python：从知识库到 Agent 的全链路开发

引言

在 2024 到 2026 年的 AI 浪潮中，大语言模型（LLM）的能力边界不断被重新定义。然而，当企业试图将 AI 引入内部业务时，往往会遇到三大痛点：数据隐私与安全——核心业务数据不能直接传给公共大模型；模型“幻觉” ——大模型会一本正经地胡说八道，这在严谨的商业场景中可能是致命的；知识滞后——通用模型不了解企业的内部规章、产品手册和实时业务数据。

为了解决这些问题，RAG（Retrieval-Augmented Generation，检索增强生成） 成为了企业 AI 落地的“黄金架构”。而更进一步，Agent（智能体） 则赋予了 LLM 自主决策和工具调用的能力，让它从“聊天机器”进化为“能干活”的数字员工。

本文将带你使用 Python 和 LangChain 框架，从零构建一个完整的 RAG 知识库系统，并在此基础上扩展出具备工具调用能力的 Agent。全文包含可运行的代码，涵盖从文档加载、文本分块、向量存储、检索生成到 Agent 循环的全链路开发。

第一部分：RAG 核心概念与架构

1.1 什么是 RAG？

简单来说，RAG 就是 “开卷考试” 。当用户提出问题时，系统不会让大模型直接凭记忆回答，而是先去企业的私有数据库（如文档、Wiki）中检索出相关的参考资料，然后将这些资料作为“上下文”喂给大模型，让其总结生成最终答案。

RAG 的核心优势在于：

• 零幻觉：答案基于检索到的真实文档，有据可查
• 低成本：无需花费巨资微调模型，只需更新知识库文档即可
• 高安全：数据留在本地向量数据库中

1.2 RAG 的三层架构

任何 RAG 系统都建立在基础流水线上，核心分为三个部分：

第二部分：环境准备与依赖安装

2.1 Python 环境

建议使用 Python 3.10 及以上版本。推荐使用虚拟环境管理项目依赖：

python -m venv rag_agent_env
source rag_agent_env/bin/activate # Linux/Mac
# 或 rag_agent_env\\Scripts\\activate # Windows

2.2 安装依赖包

我们将使用以下核心库：

pip install langchain langchain-openai langchain-community faiss-cpu python-dotenv chromadb tiktoken

各包的作用：

• langchain 系列：最流行的 LLM 应用开发框架，提供模块化组件
• faiss-cpu：Meta 开源的高性能向量数据库，适合本地部署
• chromadb：轻量级向量数据库，适合中小规模场景
• python-dotenv：安全管理 API Key
• tiktoken：用于文本分块时的 Token 计数

2.3 配置 API Key

在项目根目录创建 .env 文件，填入大模型 API 配置：

# 以 OpenAI 为例
OPENAI_API_KEY=sk-xxxxxxxxxxxxxxxxxxxxxxxx
OPENAI_BASE_URL=https://api.openai.com/v1

# 如果使用国内模型（如 DeepSeek）
# OPENAI_API_KEY=sk-xxxx
# OPENAI_BASE_URL=https://api.deepseek.com/v1

第三部分：构建知识库——RAG 核心实现

3.1 文档加载

首先，我们需要加载知识文档。这里以文本文件为例，也可以扩展支持 PDF、Markdown、网页等格式。

import os
from dotenv import load_dotenv
from langchain_community.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_openai import ChatOpenAI
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate

# 加载环境变量
load_dotenv()

# 第一步：加载文档
def load_documents(file_paths):
"""加载多个文档"""
documents = []
for path in file_paths:
loader = TextLoader(path, encoding='utf-8')
documents.extend(loader.load())
return documents

# 示例：加载公司员工手册
# docs = load_documents(["./data/employee_handbook.txt", "./data/company_policy.txt"])

3.2 文本分块

大模型的上下文窗口有限，因此需要将长文档切分为合适的文本块。我们使用 RecursiveCharacterTextSplitter，它会尝试按段落、句子等层级进行递归切分，保证语义连贯性。

def split_documents(documents, chunk_size=1000, chunk_overlap=200):
"""
将文档切分为文本块

Args:
documents: 原始文档列表
chunk_size: 每个块的大小（字符数）
chunk_overlap: 块之间的重叠字符数，保证上下文连贯
"""
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
separators=["\\n\\n", "\\n", "。", "！", "？", "；", "，", " ", ""]
)
splits = text_splitter.split_documents(documents)
print(f"文档已切分为 {len(splits)} 个文本块")
return splits

3.3 向量化与存储

将文本块通过嵌入模型转换为向量，并存入向量数据库。

def build_vectorstore(splits, embedding_model=None):
"""
构建向量数据库

Args:
splits: 切分后的文本块
embedding_model: 嵌入模型，默认使用 OpenAI 的 text-embedding-ada-002
"""
if embedding_model is None:
embeddings = OpenAIEmbeddings(model="text-embedding-ada-002")
else:
embeddings = embedding_model

# 创建 FAISS 向量数据库
vectorstore = FAISS.from_documents(splits, embeddings)

# 可选：保存到本地以便后续复用
vectorstore.save_local("./faiss_index")
print("向量数据库已构建并保存")
return vectorstore

# 完整流程：加载 -> 切分 -> 向量化
def build_knowledge_base(file_paths):
"""一站式构建知识库"""
docs = load_documents(file_paths)
splits = split_documents(docs)
vectorstore = build_vectorstore(splits)
return vectorstore

3.4 检索问答——RAG 核心链路

有了向量数据库后，我们就可以实现检索增强生成了。

def create_rag_chain(vectorstore, model_name="gpt-3.5-turbo"):
"""
创建 RAG 问答链

Args:
vectorstore: 向量数据库
model_name: 大模型名称
"""
# 创建检索器
retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

# 初始化大模型
llm = ChatOpenAI(model_name=model_name, temperature=0)

# 定义提示模板
prompt_template = """
你是一个专业的知识助手。请基于以下提供的上下文信息来回答用户的问题。
如果上下文信息不足以回答问题，请诚实地说"根据现有知识库无法回答该问题"。

<上下文>
{context}
</上下文>

用户问题：{question}

请给出准确、简洁的回答：
"""

PROMPT = PromptTemplate(
template=prompt_template,
input_variables=["context", "question"]
)

# 构建 RAG 链
qa_chain = RetrievalQA.from_chain_type(
llm=llm,
chain_type="stuff",
retriever=retriever,
chain_type_kwargs={"prompt": PROMPT},
return_source_documents=True
)

return qa_chain

def ask_question(qa_chain, question):
"""向 RAG 系统提问"""
result = qa_chain.invoke({"query": question})
print(f"问题：{question}")
print(f"回答：{result['result']}")
print(f"参考来源：{len(result['source_documents'])} 个文档片段")
return result

3.5 完整运行示例

将以上代码整合，我们可以得到一个完整可运行的 RAG 系统：

# 完整示例代码
import os
from dotenv import load_dotenv
from langchain_community.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import FAISS
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate

load_dotenv()

def main():
# 1. 准备文档（这里创建一个示例文档）
sample_doc = """
公司员工手册（2026年版）

第一章 考勤制度
公司实行弹性工作制，核心工作时间段为上午10:00至下午16:00。
员工每月有2天带薪病假，需提前在OA系统提交申请。
年假标准：入职满1年享有5天带薪年假，满3年享有10天。

第二章 薪酬福利
公司每月5日发放上月工资，如遇节假日顺延。
年终奖根据年度绩效考核结果发放，一般为1-3个月工资。
公司为所有正式员工缴纳五险一金。

第三章 培训与发展
公司每年为员工提供不少于40小时的内部培训。
员工可申请外部培训补贴，最高报销金额为5000元/年。
"""

# 保存为临时文件
with open("./sample_handbook.txt", "w", encoding="utf-8") as f:
f.write(sample_doc)

# 2. 加载文档
loader = TextLoader("./sample_handbook.txt", encoding="utf-8")
docs = loader.load()

# 3. 切分文档
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=500,
chunk_overlap=50
)
splits = text_splitter.split_documents(docs)

# 4. 构建向量数据库
embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_documents(splits, embeddings)

# 5. 创建 RAG 链
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
llm = ChatOpenAI(temperature=0)

prompt = PromptTemplate(
template="基于以下上下文回答问题：\\n{context}\\n问题：{question}\\n回答：",
input_variables=["context", "question"]
)

qa_chain = RetrievalQA.from_chain_type(
llm=llm,
retriever=retriever,
chain_type_kwargs={"prompt": prompt}
)

# 6. 测试问答
questions = [
"公司的年假政策是怎样的？",
"公司什么时候发工资？",
"员工每年有多少培训时间？"
]

for q in questions:
print(f"\\n{'='*50}")
print(f"Q: {q}")
print(f"A: {qa_chain.invoke({'query': q})['result']}")

# 清理临时文件
os.remove("./sample_handbook.txt")

if __name__ == "__main__":
main()

第四部分：从 RAG 到 Agent——赋予 AI 行动能力

4.1 什么是 Agent？

如果说 RAG 解决了 LLM“记忆”的问题，那么 Agent（智能体） 解决的是 LLM“行动”的问题。

标准聊天机器人的交互流程是：用户发送消息 → LLM 生成响应 → 流程结束。一次输入对应一次输出，无法处理多步骤的复杂任务。

而 Agent 被设计用来行动（Act），而不只是响应。两者的差异可以归结为一个基础的编程概念：while 循环。Agent 循环是一种迭代过程：LLM 在其中使用工具、根据反馈做出调整，并反复执行，直到任务彻底完成。

4.2 ReAct 模式

ReAct（Reasoning + Acting）是 Agent 最经典的工作模式。它将推理和行动交替进行：

• 推理（Reason） ：LLM 分析当前状态，思考下一步应该做什么
• 行动（Act） ：执行具体的动作——调用 API、查询数据库、运行代码等
• 观察（Observe） ：查看行动的结果，判断是否达成目标，或需要调整计划

这个循环可以简化为三步：推理 → 行动 → 观察。

4.3 工具（Tool）的定义

在 Agent 体系中，工具（Tool） 是 Agent 与外部世界交互的接口。一个工具就是一个 Python 函数，Agent 可以自主决定何时调用它、传入什么参数。你决定工具具体做什么，而模型决定何时使用它。

LangChain 提供了 @tool 装饰器，可以方便地将 Python 函数转换为 Agent 可识别的工具。

4.4 构建 RAG Agent

现在，我们将前面构建的 RAG 知识库封装成一个工具，让 Agent 可以自主决定何时查询知识库。

from langchain.agents import create_agent, AgentExecutor
from langchain.tools import tool
from langchain_openai import ChatOpenAI
from langchain.memory import ConversationBufferMemory
from langchain_core.messages import HumanMessage, AIMessage

# 假设我们已经有了一个构建好的 vectorstore
# vectorstore = build_knowledge_base(["./data/handbook.txt"])

@tool
def search_knowledge_base(query: str) –> str:
"""
在企业知识库中搜索相关信息。

Args:
query: 搜索查询，用自然语言描述你想了解的内容

Returns:
从知识库中检索到的相关信息摘要
"""
# 检索相关文档
docs = vectorstore.similarity_search(query, k=3)
if not docs:
return "未找到相关信息。"

# 组合检索结果
context = "\\n\\n".join([doc.page_content for doc in docs])
return f"检索到以下相关信息：\\n{context}"

@tool
def calculate(expression: str) –> str:
"""
执行数学计算。支持基本的四则运算。

Args:
expression: 数学表达式，如 "2 + 3 * 4"

Returns:
计算结果
"""
try:
# 安全起见，只允许数字和运算符
allowed = set("0123456789+-*/(). ")
if not all(c in allowed for c in expression):
return "错误：表达式包含不支持的字符"
result = eval(expression)
return f"计算结果：{result}"
except Exception as e:
return f"计算错误：{str(e)}"

@tool
def get_current_time() –> str:
"""获取当前日期和时间"""
from datetime import datetime
return datetime.now().strftime("%Y年%m月%d日 %H:%M:%S")

4.5 创建 Agent

有了工具之后，我们就可以创建 Agent 了。

def create_rag_agent(vectorstore):
"""创建一个具备知识库检索能力的 Agent"""

# 将 vectorstore 注入到工具函数中
global search_knowledge_base
# 这里简化处理，实际应该使用闭包或类来管理状态

# 定义工具列表
tools = [search_knowledge_base, calculate, get_current_time]

# 初始化大模型
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

# 创建 Agent
agent = create_agent(
model=llm,
tools=tools,
system_prompt="""你是一个智能助手，拥有以下能力：
1. 通过 search_knowledge_base 工具查询企业内部知识库
2. 通过 calculate 工具执行数学计算
3. 通过 get_current_time 工具获取当前时间

请根据用户的问题，自主决定使用哪些工具来完成任务。
如果需要查询知识库，请使用 search_knowledge_base 工具。
如果需要计算，请使用 calculate 工具。
在给出最终答案之前，请确保已经获取了足够的信息。
"""
)

# 创建 Agent Executor（负责管理执行循环）
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True, # 打印详细执行过程
max_iterations=5 # 最多迭代5次，防止死循环
)

return agent_executor

def chat_with_agent(agent_executor, user_input):
"""与 Agent 对话"""
result = agent_executor.invoke({"input": user_input})
return result["output"]

4.6 Agent 循环的底层原理

Agent 循环的核心逻辑可以用以下伪代码表示：

def agent_loop(messages, tools, llm, max_steps=10):
"""Agent 循环的简化实现"""
done = False
step = 0

while not done and step < max_steps:
step += 1

# 调用 LLM
response = llm.invoke(messages)

# 检查 LLM 是否请求调用工具
if hasattr(response, 'tool_calls') and response.tool_calls:
# 执行工具调用
for tool_call in response.tool_calls:
tool_name = tool_call['name']
tool_args = tool_call['args']
tool_result = execute_tool(tool_name, tool_args)

# 将工具执行结果添加到消息中
messages.append({
"role": "tool",
"tool_call_id": tool_call['id'],
"content": tool_result
})
else:
# 没有工具调用，任务完成
done = True
return response.content

return "达到最大迭代次数，任务终止"

这就是 Agent 的“推理 → 行动 → 观察”循环在代码层面的体现。

第五部分：完整实战——企业智能问答 Agent

让我们将所有组件整合在一起，构建一个完整的企业智能问答 Agent。

5.1 完整代码

"""
企业智能问答 Agent
功能：基于企业知识库的 RAG 检索 + 工具调用能力
"""

import os
from dotenv import load_dotenv
from langchain_community.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import FAISS
from langchain.tools import tool
from langchain.agents import create_agent, AgentExecutor
from datetime import datetime
import json

load_dotenv()

# ============ 第一步：构建知识库 ============
def build_knowledge_base_from_texts(texts, persist_path="./faiss_index"):
"""从文本列表构建知识库"""
# 将文本转换为 Document 对象
from langchain_core.documents import Document
documents = [Document(page_content=text) for text in texts]

# 切分
splitter = RecursiveCharacterTextSplitter(
chunk_size=500,
chunk_overlap=50
)
splits = splitter.split_documents(documents)

# 向量化并存储
embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_documents(splits, embeddings)
vectorstore.save_local(persist_path)
print(f"知识库已构建，共 {len(splits)} 个文本块")
return vectorstore

# ============ 第二步：准备知识数据 ============
knowledge_texts = [
"""
公司考勤制度：
1. 实行弹性工作制，核心工作时间 10:00-16:00
2. 每月2天带薪病假，需在OA系统提前申请
3. 年假：入职满1年5天，满3年10天，满5年15天
4. 迟到早退超过30分钟记为半天事假
""",
"""
公司薪酬福利政策：
1. 每月5日发薪，遇节假日顺延
2. 年终奖：1-3个月工资，基于绩效考核
3. 五险一金：按当地最高标准缴纳
4. 餐补：每天30元，随工资发放
5. 交通补贴：每月500元
""",
"""
员工培训与发展：
1. 每年至少40小时内部培训
2. 外部培训补贴：最高5000元/年
3. 晋升通道：专员 → 高级专员 → 主管 → 经理 → 总监
4. 每季度进行一次绩效评估
"""
]

# 构建知识库
vectorstore = build_knowledge_base_from_texts(knowledge_texts)

# ============ 第三步：定义工具 ============
@tool
def search_company_knowledge(query: str) –> str:
"""
搜索公司内部知识库，获取相关政策、制度、福利等信息。

Args:
query: 用户的查询问题，用自然语言描述

Returns:
从知识库中检索到的相关信息
"""
docs = vectorstore.similarity_search(query, k=3)
if not docs:
return "未找到相关信息，请尝试用不同方式提问。"
context = "\\n\\n—\\n\\n".join([doc.page_content for doc in docs])
return f"【知识库检索结果】\\n{context}"

@tool
def calculate(expression: str) –> str:
"""
执行数学计算。

Args:
expression: 数学表达式，如 "5000 * 1.1"
"""
try:
allowed = set("0123456789+-*/(). ")
if not all(c in allowed for c in expression):
return "错误：表达式包含不支持的字符"
result = eval(expression)
return f"计算结果：{result}"
except Exception as e:
return f"计算错误：{str(e)}"

@tool
def get_current_time() –> str:
"""获取当前日期和时间"""
return datetime.now().strftime("%Y年%m月%d日 %H:%M:%S")

# ============ 第四步：创建 Agent ============
def create_company_agent():
"""创建企业智能问答 Agent"""
tools = [search_company_knowledge, calculate, get_current_time]

llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

agent = create_agent(
model=llm,
tools=tools,
system_prompt="""你是一个企业智能助手，专门回答员工关于公司政策、制度、福利等方面的问题。

工作原则：
1. 对于公司政策类问题（考勤、薪酬、福利、培训等），必须使用 search_company_knowledge 工具检索知识库
2. 对于计算类问题，使用 calculate 工具
3. 对于时间类问题，使用 get_current_time 工具
4. 如果知识库中没有相关信息，请诚实告知用户
5. 回答要准确、专业、友好

请根据用户的问题，自主决定调用哪些工具。
"""
)

agent_executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True,
max_iterations=5
)

return agent_executor

# ============ 第五步：运行 ============
def main():
print("=" * 60)
print("欢迎使用企业智能问答 Agent")
print("你可以询问公司政策、进行计算、查询时间等")
print("输入 'quit' 或 'exit' 退出")
print("=" * 60)

agent = create_company_agent()

while True:
user_input = input("\\n你：").strip()
if not user_input:
continue
if user_input.lower() in ['quit', 'exit', 'bye']:
print("再见！")
break

try:
result = agent.invoke({"input": user_input})
print(f"\\n助手：{result['output']}")
except Exception as e:
print(f"\\n错误：{str(e)}")

if __name__ == "__main__":
main()

5.2 运行效果示例

============================================================
欢迎使用企业智能问答 Agent
你可以询问公司政策、进行计算、查询时间等
输入 'quit' 或 'exit' 退出
============================================================

你：公司的年假政策是什么？

> Entering new AgentExecutor chain…
> 思考：用户询问年假政策，需要查询知识库
> 调用 search_company_knowledge("年假政策")
> 观察：【知识库检索结果】
公司考勤制度：…
> 思考：已获取足够信息，可以回答
> 最终回答：

助手：根据公司考勤制度，年假政策如下：
– 入职满1年：享有5天带薪年假
– 入职满3年：享有10天带薪年假
– 入职满5年：享有15天带薪年假

你：我入职2年了，可以休几天？

> 思考：用户入职2年，属于满1年不满3年的情况
> 调用 search_company_knowledge("年假")
> 观察：…
> 思考：用户入职2年，适用"满1年5天"的规定

助手：根据公司制度，您入职2年，属于"入职满1年"的范畴，享有 **5天** 带薪年假。

你：5天年假折算成小时是多少？（按8小时工作制）

> 思考：需要计算 5 * 8
> 调用 calculate("5 * 8")
> 观察：计算结果：40

助手：5天年假按每天8小时计算，共计 **40小时**。

第六部分：进阶优化与生产实践

6.1 检索优化

在实际生产环境中，基础 RAG 可能面临检索精度不足的问题。以下是一些优化方向：

• 查询重写（Query Rewriting） ：将用户问题改写为更适合检索的形式
• 多路检索（Multi-path Retrieval） ：同时使用向量检索和关键词检索（BM25），合并结果
• 重排序（Re-ranking） ：对初步检索结果进行精细排序，将最相关的放在前面
• HyDE（Hypothetical Document Embeddings） ：让 LLM 先生成一个假设性的答案，再用这个答案去检索

6.2 Agent 的健壮性

• 设置最大迭代次数：防止 Agent 陷入死循环
• 错误处理：工具调用失败时，Agent 应能优雅降级
• 记忆机制：使用 ConversationBufferMemory 或 InMemorySaver 实现对话记忆

6.3 向量数据库选型

6.4 模型选择

本文使用的是 OpenAI 的模型，但你可以无缝切换到其他模型：

• 国内模型：DeepSeek、通义千问（Qwen）、智谱（GLM）
• 本地模型：通过 Ollama 运行 Qwen、Llama 等开源模型
• 只需修改 base_url 和 api_key 即可切换

总结

本文从零开始，使用 Python 和 LangChain 构建了一个完整的 RAG + Agent 系统。我们走过了从知识库构建到智能体创建的全链路：

RAG 解决了 LLM 知识滞后和幻觉的问题，而 Agent 则赋予了 LLM 自主决策和行动的能力。两者的结合，让大模型从“能聊”走向了“能干”。

这套架构已经在越来越多的企业中得到验证——从 HR 政策问答、产品技术支持到数据分析助手。希望本文的代码和思路能够成为你探索 AI 应用开发的起点。