本文实现了一个基于 RAG（检索增强生成）的智能体，专门用于回答关于 “2024 年股市表现” 的问题。它通过加载指定的 PDF 文档（Stock_Market_Performance_2024.pdf），构建向量数据库，并结合 LLM（大语言模型）生成基于文档内容的精准回答。

一、核心流程如下：

1. 初始化与依赖加载

• 导入必要库：langgraph（状态图构建）、langchain（LLM、工具、文档处理）、chroma（向量数据库）等。
• 通过load_dotenv()加载环境变量（主要是 API 密钥）。

2. 文档处理与向量库构建

• PDF 加载：使用PyPDFLoader加载Stock_Market_Performance_2024.pdf，并验证文件是否存在。
• 文档分割：通过RecursiveCharacterTextSplitter将 PDF 按chunk_size=1000（每块 1000 字符）、chunk_overlap=200（重叠 200 字符）分割为小片段，便于后续检索。
• 向量存储：使用OpenAIEmbeddings（OpenAI 的嵌入模型）将分割后的文档片段转换为向量，存储在 Chroma 向量数据库中（路径为persist_directory，集合名为stock_market）。

3. 工具与智能体定义

• 检索工具（retriever_tool）：封装向量库的检索功能，当用户提问时，通过retriever.invoke(query)从向量库中获取最相关的 5 个文档片段（search_kwargs={"k":5}），并返回结果。
• LLM 配置：使用deepseek-chat模型作为主 LLM，并通过bind_tools(tools)绑定检索工具，使模型能生成工具调用指令。
• 状态定义（AgentState）：通过TypedDict定义状态结构，核心字段messages用于存储对话历史（用户消息、LLM 响应、工具结果等），通过add_messages确保消息追加而非替换。

4. 状态图（LangGraph）工作流程

构建了包含两个节点的状态图，形成 “LLM 决策→工具检索→LLM 生成回答” 的闭环：

• llm节点（call_llm函数）：接收当前对话历史，结合系统提示（system_prompt）调用 LLM 生成响应（可能包含工具调用指令）。
• retriever_agent节点（take_action函数）：解析 LLM 生成的工具调用指令，执行retriever_tool获取相关文档片段，将结果包装为ToolMessage返回。
• 流转逻辑：
  • llm节点后通过should_continue判断：若 LLM 响应包含工具调用，则流向retriever_agent节点执行检索；否则直接结束（END）。
  • retriever_agent节点执行后，结果返回llm节点，继续生成回答（可多轮检索）。

5. 交互流程

通过running_agent函数实现用户交互：

• 循环接收用户提问（输入 “exit” 或 “quit” 退出）。
• 将用户问题转换为HumanMessage，传入智能体。
• 智能体通过多轮 “检索→生成” 生成回答，并打印结果。

二、代码中使用了两个需要 API 密钥的服务，需在.env文件中配置：

1. 智普AI API 密钥

   • 用途：代码中使用

     embeddings = ZhipuAIEmbeddings(
         model="embedding-3",
         api_key=os.getenv("ZHIPUAI_API_KEY")  # 读取.env中的ZHIPUAI_API_KEY
     )

     生成文档嵌入向量，这依赖 智普AI的嵌入模型服务。
   • 环境变量名：ZHIPUAI_API_KEY
   • 获取方式：登录智普AI平台，创建 API 密钥。

2. DeepSeek API 密钥

   • 用途：代码中通过init_chat_model("deepseek-chat")使用 DeepSeek 的deepseek-chat模型作为主 LLM，需要 DeepSeek 的 API 密钥。
   • 环境变量名：DEEPSEEK_API_KEY（具体以 DeepSeek 官方文档为准，部分平台可能使用其他名称，如API_KEY）
   • 获取方式：登录DeepSeek 平台，创建 API 密钥。

三、代码分析

1、加载依赖包

from dotenv import load_dotenv
import os
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, Sequence
from langchain_core.messages import BaseMessage, SystemMessage, HumanMessage, ToolMessage
from operator import add as add_messages
from langchain.chat_models import init_chat_model
# 用于加载 PDF 格式的文档，并将其解析为 LangChain 可处理的Document对象（包含文本内容和元数据），每页对应一个Document实例
from langchain_community.document_loaders import PyPDFLoader
# 将长文本递归地分割为较小的文本片段（chunk）
from langchain.text_splitter import RecursiveCharacterTextSplitter
#集成智谱 AI（ZhipuAI）的嵌入模型（如embedding-3），用于将文本转换为向量（Embedding）
from langchain_community.embeddings import ZhipuAIEmbeddings
# LangChain 与 Chroma 向量数据库的集成接口，用于将文本片段的嵌入向量（Embedding）存储到 Chroma 中
from langchain_chroma import Chroma
from langchain_core.tools import tool

2. 准备工作

在项目根据录下，新建.env文件，在文件中设置API_KEY

ZHIPUAI_API_KEY= 
DEEPSEEK_API_KEY= 

加载环境变量、大模型

#加载环境变量
load_dotenv()
# 初始化大模型
llm = init_chat_model(
    model="deepseek-chat",
    temperature=0) 
# 初始化词嵌入向量化
embeddings = ZhipuAIEmbeddings(
    model="embedding-3",
    api_key=os.getenv("ZHIPUAI_API_KEY")  # 读取.env中的ZHIPUAI_API_KEY
)

加载pdf文件

pdf_path = "Stock_Market_Performance_2024.pdf"

if not os.path.exists(pdf_path):
    raise FileNotFoundError(f"PDF file not found: {pdf_path}")

pdf_loader = PyPDFLoader(pdf_path)

#判断pdf文件是否加载成功
try:
    pages = pdf_loader.load()
    print(f"PDF has been loaded and has {len(pages)} pages")
except Exception as e:
    print(f"Error loading PDF: {e}")
    raise

3. 文档分割和向量化、存储

文档分割、向量化、向量存储是 RAG（检索增强生成）流程中 “文档预处理” 和 “向量存储” 环节的核心步骤。

分割：

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200
)
pages_split = text_splitter.split_documents(pages)

1. 核心功能

• RecursiveCharacterTextSplitter 是 LangChain 中最常用的文本分割工具，用于将长文档（这里是 PDF 的pages）递归分割为更小的文本片段（chunk），解决以下问题：
  • 大模型有上下文长度限制（如单次输入不能超过几万字符），长文档无法直接处理。
  • 嵌入模型（生成文本向量的模型）对长文本的语义捕捉能力有限，短片段的向量更能精准反映局部语义。
  • 检索时，短片段能提高匹配精度（用户问题通常只与文档中的部分内容相关）。

2. 关键参数解析

• chunk_size=1000：
  每个文本片段的字符数上限（注意：不是单词数，中文、英文、符号均按字符计数）。这里设置为 1000，意味着每个片段约 1000 字符（中文约 300-500 字，英文约 200-300 词）。

  • 数值过小：片段过碎，可能破坏语义完整性（如拆分一个完整的句子或段落）。
  • 数值过大：可能超过模型上下文限制，检索精度下降。

• chunk_overlap=200：
  相邻两个片段的重叠字符数。这里设置为 200，意味着后一个片段会重复前一个片段末尾的 200 字符。

  • 作用：避免因分割导致的语义断裂。例如，一个跨页的重要概念，通过重叠部分能同时出现在两个片段中，确保检索时不丢失信息。
  • 通常设置为 chunk_size 的 10%-20%（这里 200 是 1000 的 20%，属于合理范围）。

3.split_documents(pages) 的效果

• pages 是通过PyPDFLoader加载的 PDF 页面列表（每个元素是一个Document对象，包含单页文本和元数据）。
• split_documents 会遍历pages中的每个页面，按上述参数分割为多个小片段，最终返回一个包含所有片段的列表pages_split（每个片段仍是Document对象，保留原始元数据如页码）。

向量库配置：存储路径与集合名

#这里是 Windows 路径，r表示原始字符串，避免转义字符问题
persist_directory = r"C:\Users\40258\LangGraph-Course-freeCodeCamp\Agents"
collection_name = "stock_market"
if not os.path.exists(persist_directory):
    os.makedirs(persist_directory)

Chroma 会将文本片段的向量、元数据等信息持久化到persist_directory目录，下次运行程序时可直接加载，无需重新处理文档（节省时间和计算资源）。

• 注意：需确保该路径有写入权限，否则会出现存储失败的错误。

collection_name：向量集合的名称，类似数据库中的 “表名”。

当一个存储目录下需要存储多个不同文档的向量时（如同时处理 “股市报告” 和 “产品手册”），可通过不同collection_name区分，避免数据混淆。

将分割后的文档片段转换为向量并存储到 Chroma 向量数据库中

try:
    # 创建向量库
    vectorstore = Chroma.from_documents(
        documents=pages_split,  # 待转换为向量的文档片段列表
        embedding=embeddings,   # 用于生成向量的嵌入模型（如ZhipuAIEmbeddings）
        persist_directory=persist_directory,  # 向量库本地存储路径
        collection_name=collection_name       # 向量集合名称（类似数据库表名）
    )

except Exception as e:
    print(f"Error setting up ChromaDB: {str(e)}")  # 打印错误信息
    raise  # 重新抛出异常，终止程序或让上层处理

• persist_directory：指定向量数据的本地存储路径，Chroma 会在此路径下生成chroma.sqlite3等文件，持久化存储向量和元数据。

# 创建检索器，检索方式similarity,每次返回5个块
retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 5}  # K is the amount of chunks to return
)

接下啦就进入了LangGraph的主要流程了，分别为定义状态字典信息类、加载问答大模型、定义工具，给大模型绑定工作功能。

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add_messages]



@tool
def retriever_tool(query: str) -> str:
    """
    这个工具从Stock Market Performance 2024 document文档中检索信息，并返回检索的信息
    """

    docs = retriever.invoke(query) 

    if not docs:
        return "I found no relevant information in the Stock Market Performance 2024 document."

    results = []
    for i, doc in enumerate(docs):
        results.append(f"Document {i + 1}:\n{doc.page_content}")

    return "\n\n".join(results)

tools = [retriever_tool]

llm = init_chat_model("deepseek-chat")
llm = llm.bind_tools(tools)

定义节点函数、创建graph_builder、给graph_builder添加节点和边，编译graph_builder得到智能体实体，智能体通过invoke或stream方法触发系统按照graph_builder中创建的工作流程（图）进行工作。

def call_llm(state: AgentState) -> AgentState:
    """调用大模型llm."""
    messages = list(state['messages'])
    messages = [SystemMessage(content=system_prompt)] + messages
    message = llm.invoke(messages)
    return {'messages': [message]}

下面这个名为take_action的函数，是 LangGraph 状态图中负责执行工具调用的核心节点逻辑。它的主要功能是解析大模型（LLM）生成的工具调用指令，执行对应的工具，并将结果封装成消息格式工具消息。

tools_dict = {t.name: t for t in tools} #得到工具名:工具的字典

def take_action(state: AgentState) -> AgentState:
    """
    执行LLM响应中包含的工具调用指令
    作用：解析大模型生成的工具调用请求，执行对应的工具（如检索器），
          将执行结果封装为标准消息格式并返回更新智能体状态
    """

    # 从当前状态的最后一条消息中提取工具调用指令
    # 最后一条消息通常是LLM返回的AIMessage，其中可能包含tool_calls字段
    tool_calls = state['messages'][-1].tool_calls
    
    # 存储所有工具调用的结果（ToolMessage对象）
    results = []
    
    # 遍历每一个工具调用指令，依次执行
    for t in tool_calls:
        # 打印当前调用的工具名称和参数（调试用，方便跟踪流程）
        # t['args'].get('query')获取查询参数，若不存在则显示默认文本
        print(f"调用工具: {t['name']}，查询内容: {t['args'].get('query', '未提供查询内容')}")

        # 检查工具是否存在于工具字典（tools_dict）中
        if not t['name'] in tools_dict:
            # 工具不存在时，打印错误信息并生成错误提示
            print(f"\n工具: {t['name']} 不存在")
            result = "工具名称错误，请重试并从可用工具列表中选择"

        else:
            # 工具存在时，调用工具的invoke方法执行具体功能
            # 传入查询参数（默认为空字符串）
            result = tools_dict[t['name']].invoke(t['args'].get('query', ''))
            # 打印结果长度（调试用，快速判断是否返回有效内容）
            print(f"结果长度: {len(str(result))}")

        # 将工具执行结果封装为ToolMessage对象（LangChain的标准消息类型）
        # - tool_call_id: 与调用指令的ID关联，确保上下文对应
        # - name: 工具名称
        # - content: 工具执行结果（转换为字符串）
        results.append(ToolMessage(
            tool_call_id=t['id'],
            name=t['name'],
            content=str(result)
        ))

    # 打印工具执行完成的提示（调试用）
    print("工具执行完成，返回给模型继续处理！")
    
    # 将工具结果列表作为新的消息，更新智能体状态并返回
    return {'messages': results}

当大模型（如 deepseek-chat）决定调用工具时，会生成结构化的工具调用指令（存储在 tool_calls 中），格式类似：

[
    {
        "name": "retriever_tool",  # 工具名称
        "args": {"query": "2024年美股涨幅是多少？"},  # 工具参数，包含query
        "id": "call_123"  # 调用唯一ID
    }
]

比如：t[args]得到了

{"query": "2024年股市表现如何？"}

再通过.get("query","xxx") 可以得到“ 2024年美股涨幅是多少？”，这里的xxx表示如果没有得到结果，就用xxx代替。

条件决策函数

def should_continue(state: AgentState):
    """如果最后一条消息中有tool_calls属性，就返回True."""
    result = state['messages'][-1]
    return hasattr(result, 'tool_calls') and len(result.tool_calls) > 0

定义图

raph = StateGraph(AgentState)
graph.add_node("llm", call_llm)
graph.add_node("retriever_agent", take_action)

graph.add_conditional_edges(
    "llm",
    should_continue,
    {True: "retriever_agent", False: END}
)
graph.add_edge("retriever_agent", "llm")
graph.set_entry_point("llm")

rag_agent = graph.compile()

运行智能体

def running_agent():
    print("\n=== RAG AGENT===")

    while True:
        user_input = input("\n请输入你的问题: ")
        if user_input.lower() in ['exit', 'quit']:
            break

        messages = [HumanMessage(content=user_input)]  # converts back to a HumanMessage type

        result = rag_agent.invoke({"messages": messages})

        print("\n=== ANSWER ===")
        print(result['messages'][-1].content)


running_agent()