从零构建自主决策的Agentic RAG系统

1. 项目概述：构建一个具备自主思考能力的Agentic RAG系统

在当今AI技术快速发展的背景下，传统的检索增强生成(RAG)系统已经不能满足复杂场景的需求。我们需要的是一种能够像人类一样思考、能够自主决定何时检索、如何检索，并能连续调用不同工具解决问题的智能系统。这就是Agentic RAG系统的核心价值所在。

这个项目将带你从零开始构建一个完整的Agentic RAG系统，它具备以下关键能力：

• 能够根据问题类型自动判断是否需要检索
• 可以连续调用不同的工具（如向量数据库检索和网络搜索）来获取信息
• 具备类似人类的"思考-行动-评估"循环机制
• 能够处理公司内部知识和外部实时信息

我们将使用LangGraph来构建这个系统的核心逻辑框架，它特别适合处理这种需要多步骤决策的场景。整个系统将包含自定义工具开发、状态管理、条件逻辑路由等关键组件，最终形成一个能够自主决策的智能体。

2. 环境准备与工具配置

2.1 安装必要的Python包

首先，我们需要安装项目依赖的各种Python包。打开终端或命令行界面，执行以下安装命令：

bash复制pip install sentence_transformers
pip install langgraph==1.0.7
pip install langchain_openai==1.1.7
pip install langchain-core==1.2.7
pip install google-search-results

这些包各自的作用如下：

• sentence_transformers：用于文本嵌入生成，将文本转换为向量表示
• langgraph：构建智能体工作流的核心框架
• langchain_openai：提供与OpenAI模型的接口
• langchain_core：LangChain的核心组件
• google-search-results：用于实现网络搜索功能

注意：在实际部署时，建议使用虚拟环境来管理这些依赖，以避免包版本冲突。可以使用python -m venv venv创建虚拟环境，然后激活它再安装这些包。

2.2 配置API密钥

我们的系统需要使用Serper API进行网络搜索。首先需要在Serper官网注册账号并获取API密钥，然后将其设置为环境变量：

python复制import os
os.environ["SERPER_API_KEY"] = "your_actual_api_key_here"

提示：Serper提供有限的免费查询额度，适合开发和测试使用。在生产环境中，可能需要考虑升级到付费计划。

2.3 初始化语言模型

我们将使用OpenAI的语言模型作为系统的核心推理引擎：

python复制from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
    model="gpt-3.5-turbo",  # 也可以使用gpt-4等更强大的模型
    api_key="your_openai_api_key",
    temperature=0  # 设置为0以获得更确定性的输出
)

3. 构建知识库与工具系统

3.1 创建公司向量数据库

首先，我们需要建立一个包含公司信息的向量数据库。这个数据库将存储公司的各种信息，供RAG工具检索使用。

python复制from sentence_transformers import SentenceTransformer
import numpy as np

# 初始化嵌入模型
model = SentenceTransformer('all-MiniLM-L6-v2')

class VectorDatabase:
    def __init__(self):
        self.data = []
        self.embeddings = []
    
    def add(self, sentences: list):
        """向数据库中添加句子并生成对应的嵌入"""
        for sentence in sentences:
            embedding = model.encode(sentence)
            self.data.append({
                "sentence": sentence,
                "embedding": embedding
            })
            self.embeddings.append(embedding)
        self.embeddings = np.array(self.embeddings)
    
    def search(self, query_embedding, top_k=5):
        """搜索与查询最相似的前k个句子"""
        # 计算余弦相似度
        similarities = np.dot(self.embeddings, query_embedding) / (
            np.linalg.norm(self.embeddings, axis=1) * np.linalg.norm(query_embedding)
        )
        # 获取相似度最高的k个结果的索引
        top_indices = np.argsort(similarities)[-top_k:][::-1]
        # 返回结果
        return [{
            "metadata": self.data[i],
            "score": float(similarities[i])
        } for i in top_indices]

# 初始化向量数据库并添加公司信息
company_db = VectorDatabase()
company_information = [
    "Q