首页 / 文章 / 在FastAPI中嵌入LangChain Agent：实现工具功能、手册搜索以及流式处理。

Python FastAPI LangGraph RAG AI Agents ChromaDB

在FastAPI中嵌入LangChain Agent：实现工具功能、手册搜索以及流式处理。

使用FastAPI和LangChain构建应用内助手：将ChromaDB中的PDF手册作为工具提供，支持用户专属上下文、历史记录检查点以及流式回复功能。

6913 词

当一个应用程序的界面数量超过少数几个时，其文档就会变得日益繁杂，用户也会不再阅读它们。一份包含20页内容、介绍功能、配置及规则的手册固然有价值，但前提是人们无需费力搜索就能从中找到答案。内置在产品中的助手可以填补这一空白：它能根据手册回答“这是如何工作的？”这类问题，也能依据应用程序自身的数据回答“我的项目中有什么？”这类问题。

本指南将介绍这样一个助手的精简且可运行的版本。您需要将 LangChain 智能体集成到 FastAPI 服务中，为其配置一个用于读取应用数据的工具以及另一个用于在 ChromaDB 中搜索 PDF 手册的工具；根据请求将经过身份验证的用户信息传递给智能体，利用 LangGraph checkpointer 保存对话历史记录，并将答案实时返回给客户端。在此过程中，我们会指出该最小化代码中存在的缺陷以及需要在正式运行前修复的问题，还会说明在投入生产前需要做哪些调整。

应用场景与相关组件

想象有一个团队正在开发用于设计光伏系统的工具。该产品最初规模较小，后来逐渐增加了电池板、逆变器、生产估算数据、屋顶布局图以及大量设计规则。如今其配套手册已有超过 20 页。不断有两大类问题出现：

关于产品本身的问题：某个功能的用途、某项设置的所在位置、适用哪条规则。这些问题的答案都在文档中。
关于用户自身工作的问题：他们选择了哪种逆变器、系统能产生多少电力、使用了哪些屋顶。这些答案存储在应用程序的数据库中，默认情况下没有模型能够知晓。

检索增强生成（RAG）技术用于处理第一类问题：对手册进行索引，当有问题出现时查找相关段落，并将其作为上下文提供给模型。工具则用于处理第二类问题：模型可以调用这些小型函数从应用程序中获取数据。将两者结合，就能得到一个既能解释产品功能又能针对具体项目进行分析的助手。

这一场景背后的实际系统拥有更多工具和更复杂的领域逻辑。下文有意简化为核心部分，以便清晰展示架构结构：

FastAPI 提供 HTTP API 并识别调用者身份。
LangChain 的智能体（运行在 LangGraph 上） 负责处理推理循环和对话状态。
大语言模型 解析每个请求，并判断是否需要外部信息。
工具模块 为智能体提供对应用功能的受控访问权限。
RAG 允许智能体检索文档内容。
ChromaDB 用于存储手册片段并执行向量搜索。
流式处理 在答案生成过程中向客户端持续发送标记数据。

这种架构的优势在于，助手运行在拥有真实用户和真实数据的现有全栈应用内部，而非作为普通聊天机器人独立存在。

项目结构

每个功能模块都有独立的包：HTTP路由、身份验证、智能体逻辑、工具以及RAG处理流程。这样的设计使得每个文件规模较小，同时也能清晰地判断新功能应放置于何处。

project/
│
├── main.py
├── .env
├── .gitignore
│
├── auth/
│   ├── __init__.py
│   └── dependencies.py
│
├── routers/
│   ├── __init__.py
│   └── chat.py
│
├── llm/
│   ├── __init__.py
│   ├── agent.py
│   ├── context.py
│   ├── orchestrator.py
│   ├── prompts.py
│   ├── provider.py
│   │
│   ├── tools/
│   │   ├── __init__.py
│   │   ├── demo_tool.py
│   │   └── manual_tool.py
│   │
│   └── rag/
│       ├── __init__.py
│       ├── config.py
│       ├── context.py
│       │
│       ├── ingestion/
│       │   ├── __init__.py
│       │   ├── loader.py
│       │   ├── chunker.py
│       │   ├── chroma.py
│       │   └── indexer.py
│       │
│       └── retrieval/
│           ├── __init__.py
│           └── retriever.py
│
├── scripts/
│   ├── __init__.py
│   └── index_manual.py
│
├── docs/
│   └── manual.pdf
│
└── chroma_data/ (*generated locally, not commited or deployed)

各部分的用途：

main.py用于构建FastAPI应用。
auth/目录存放模拟的身份验证依赖项。
routers/目录包含HTTP接口。
llm/目录存放与智能体相关的所有内容。
llm/tools/目录存放智能体可调用的函数。

llm/rag/目录中存放检索流程，分为ingestion/（加载、分割PDF并建立索引）和retrieval/（向ChromaDB发起查询）两部分。

scripts/目录包含需手动执行的命令，例如为手册建立索引。

docs/目录存放原始PDF文件。

chroma_data/目录中的数据是在本地生成的，绝不可被提交或部署。

无需一次性创建所有这些目录。构建顺序为：API层、模型、工具、RAG流程，最后是上下文管理、流式处理及历史记录功能。

第一步：带有模拟用户的FastAPI框架

首先安装项目所需的所有组件，包括Web服务器、LangChain与LangGraph、兼容OpenAI的聊天客户端、ChromaDB、PDF加载器与文本分割工具，以及用于配置管理的python-dotenv库。

pip install fastapi uvicorn langchain langgraph langchain-openai chromadb langchain-community langchain-text-splitters pypdf python-dotenv

该模型将通过OpenRouter访问，因此需要在项目根目录下创建一个包含API密钥的.env文件。

OPENROUTER_API_KEY=your_api_key_here

应立即将.env文件添加到.gitignore中。一旦密钥被纳入版本控制，就应视为已泄露。

应用程序入口点

main.py文件应保持简洁。它仅负责创建应用程序并注册聊天路由器；模型或智能体的相关代码不应放在这里。

from fastapi import FastAPI
from routers.chat import chat_router

app = FastAPI(
    title="AI Agent Demo",
)
app.include_router(chat_router)

第一个聊天端点

在routers/chat.py文件中，使用/chat前缀定义一个包含单个POST路由的路由器。目前它仅会回显传入的消息，这足以在引入任何人工智能之前确认系统连接正常。

from fastapi import APIRouter

chat_router = APIRouter(
    prefix="/chat",
    tags=["Chat"],
)

@chat_router.post("")
def ask_ai(
    message: str,
):
    return {
        "message": message,
    }

请注意，在没有请求体模型的 POST 路由中，message: str 会使得 FastAPI 从查询字符串中读取该参数。这在 Swagger UI 中进行测试时很方便，但对于实际客户端而言，通常应使用 Pydantic 模型定义的 JSON 请求体，因为查询字符串会出现在访问日志中，且存在实际长度限制。

模拟认证依赖项

在生产环境中，应用程序会验证会话 Cookie 或 JWT 并从数据库中加载用户信息。而在这里，自定义请求头可以替代所有这些功能。创建 auth/dependencies.py 文件，其中包含一个简单的 MockUser 数据类以及一个 get_current_user 函数，该函数用于读取 X-Demo-User 请求头，若缺失则返回 401 错误以拒绝请求。

from dataclasses import dataclass
from fastapi import Header, HTTPException

@dataclass
class MockUser:
    id: str
    name: str

def get_current_user(
    x_demo_user: str | None = Header(default=None),
) -> MockUser:
    if x_demo_user is None:
        raise HTTPException(
            status_code=401,
            detail="Missing X-Demo-User header",
        )
    return MockUser(
        id=x_demo_user,
        name=x_demo_user,
    )

FastAPI 的依赖注入功能现在可直接将用户传递给对应端点。只需用 Depends(get_current_user) 声明参数即可。

from fastapi import APIRouter, Depends
from auth.dependencies import (
    MockUser,
    get_current_user,
)

chat_router = APIRouter(
    prefix="/chat",
    tags=["Chat"],
)

@chat_router.post("")
def ask_ai(
    message: str,
    current_user: MockUser = Depends(
        get_current_user,
    ),
):
    return {
        "user": current_user.name,
        "message": message,
    }

客户端通过发送类似这样的请求头来标识自身：

X-Demo-User: user-123

在每个请求中，FastAPI 首先执行 get_current_user()，然后将得到的 MockUser 对象传递给 ask_ai。其核心设计思路在于职责划分：FastAPI 负责身份验证，而 AI 层只需接收一个可信任的用户对象。之后，正是这个用户对象让相关工具能够返回属于对应用户的数据。若后续将模拟认证替换为真实认证，也仅会改变这一处依赖关系。

步骤 2：通过 OpenRouter 连接模型

既然已有可用的接口端点以及已知的调用方，该服务就需要一个模型。OpenRouter提供了兼容OpenAI的API，因此只要将base_url设置为OpenRouter并传入对应的密钥，LangChain的ChatOpenAI类即可正常使用。

将该代码放入llm/provider.py文件中。它会读取.env文件，若密钥缺失会立即抛出明确的错误信息；同时会将密钥封装在Pydantic的SecretStr类中，以避免其在日志或对象表示中被意外输出。

import os
from dotenv import load_dotenv
from pydantic import SecretStr
from langchain_openai import ChatOpenAI

load_dotenv()

api_key = os.getenv(
    "OPENROUTER_API_KEY",
)
if not api_key:
    raise RuntimeError(
        "OPENROUTER_API_KEY environment variable is not set."
    )

model = ChatOpenAI(
    model="YOUR_MODEL",
    api_key=SecretStr(api_key),
    base_url="https://openrouter.ai/api/v1",
)

由于密钥来自环境变量，因此不会出现在源代码中。此时虽然已经可以向模型发送提示并获取回复，但这只是普通的LLM调用方式。我们的目标是创建一个能够自行判断何时需要使用工具的智能体。

选择模型

model 参数仅是一个 OpenRouter 模型标识符，因此无需修改其他代码即可更换模型。在比较不同候选模型时，请检查以下方面：

代理所依赖的工具调用支持情况；
流式处理支持能力；
上下文窗口的大小；
速率限制；
是否有免费版本。

OpenRouter 提供了一些免费的模型，非常适合用于实验。该模型列表会定期更新，因此请查看当前列表并筛选免费模型，而非依赖固定的推荐内容。无论选择哪种模型，都会直接被传入构造函数中：

model = ChatOpenAI(
    model="YOUR_MODEL_ID",
    api_key=SecretStr(api_key),
    base_url="https://openrouter.ai/api/v1",
)

例如，如果列表中出现了如下所示的标识符（此为撰写本文时的示例，可能已不再可用），则需将 exactly 这个字符串作为 model 传入：

google/gemma-4-26b-a4b-it:free

请记住，免费模型使用的是共享资源。它们可能会遇到速率限制或在演示最关键的时刻无法使用。切换到其他模型，或通过OpenRouter添加自己的提供商密钥，通常可以解决这个问题。在正式环境中选择模型时，应综合考虑可靠性、功能、延迟和成本，而不仅仅是价格。

步骤3：从模型到智能体

直接的模型调用只需一步：用户的文本输入后，立即得到回复。而智能体则加入了决策循环——模型先查看请求，判断是否能立即回答或需要先获取某些信息，必要时调用工具，读取结果后才能生成最终答案。大致流程如下：

简单调用：用户输入 → 大语言模型处理 → 返回响应；

代理：首先是用户，接着是代理，然后LLM决定需要什么，必要时进行工具调用或信息检索，最后生成响应。

代理模块

llm/agent.py中，LangChain的create_agent函数会根据模型和系统提示来构建代理。其内部会生成一个LangGraph图结构，从而为你自动执行模型与工具的循环处理。

from langchain.agents import create_agent
from llm.provider import model
from llm.prompts import SYSTEM_PROMPT

agent = create_agent(
    model=model,
    system_prompt=SYSTEM_PROMPT,
)

这个代理目前还没有任何工具，因此其行为很像单纯的模型。首先需要给它下达指令。

系统提示

llm/prompts.py中包含一段简短的提示语，用于告知模型它的功能、禁止编造数据，并说明不同类型的问题应对应哪种查询方式。

SYSTEM_PROMPT = """
You are an AI assistant for our demo application.
You help users understand the application and navigate the system.
Never invent data.
When information about the demo system
is required, use the available application tools.
When answering questions about the application,
use the documentation search tool.
Always answer in clear, conversational language.
""".strip()

该提示语设定了两个信息来源：

应用数据（即用户账户中的内容）来自应用工具；
应用知识（产品的工作原理）来自文档搜索。

在继续之前需要澄清一点：下面会分别介绍工具和RAG，这样更便于理解，但在最终的智能体中，文档搜索本身也是一种工具。并不存在第二种机制：智能体会看到一组可调用的函数，而搜索手册就是其中的其中之一。

第4步：第一个工具

工具是智能体被允许调用的函数。这正是该架构能够扩展的原因：无需将所有的应用数据都放入提示词中，只需暴露特定的操作，让模型在需要时再请求这些操作。

在演示中，llm/tools/demo_tool.py定义了一个工具，用于返回固定格式的项目信息。

from langchain.tools import tool

@tool
def get_my_demo_data() -> str:
    """
    Return information about the demonstration data.
    This is just for demo data. But in production, make a more detailed instruction.
    """

    return """
    Project: Aperture Analytics Dashboard
    Owner: Jordan Lee
    Status: In Progress
    Team size: 6
    Budget: $84,000
    Deadline: 2026-11-15
    Description: An internal dashboard for visualizing customer usage
    metrics, built with FastAPI and React, integrating with the
    company's data warehouse.
    """.strip()

这里有两个要点需要注意。@tool装饰器会将普通的Python函数转换为LangChain工具，其名称和参数结构均来自函数的签名。而文档字符串则会成为该工具的描述，模型在决定是否调用该工具时会读取这部分内容。在真实的系统中，这类描述需要格外重视：必须明确说明工具会返回什么结果、在何种情况下适用以及何时不适用。模糊的描述是导致智能体调用错误工具或根本不调用工具的最常见原因之一。

可通过将工具传递给create_agent来注册它：

from langchain.agents import create_agent
from llm.provider import model
from llm.prompts import SYSTEM_PROMPT
from llm.tools.demo_tool import get_my_demo_data

agent = create_agent(
    model=model,
    tools=[
        get_my_demo_data,
    ],
    system_prompt=SYSTEM_PROMPT,
)

你的代码从未决定函数何时运行。如果用户询问“我有哪些演示数据？”，模型会意识到需要账户特定信息，于是调用get_my_demo_data()。而如果用户询问“什么是演示？”，则无需查询，模型可直接回答。这种选择会在每次交互、即代理循环内部产生。

第5步：为手册构建RAG流程

现在代理虽然能够获取应用程序数据，但对产品的运作方式仍一无所知。将20页的手册内容粘贴到系统提示词中，会在每次请求时浪费令牌，且难以保持内容更新。RAG解决了这两个问题。

有必要明确界定RAG的范畴与边界。该技术并不会对文档进行任何训练或微调。在接收到查询时，系统会从手册中搜索与问题最相关的段落，并将这些段落作为上下文提供给模型，模型再基于这些内容给出答案。

整个流程分为两个阶段：

数据摄取，这一阶段与网页应用独立运行：加载PDF文件，将其分割成多个片段，并将这些片段及其嵌入向量存储在ChromaDB中。
信息检索，这一阶段在处理请求时执行：获取查询内容，在ChromaDB中搜索，挑选出最相关的片段，然后将它们传递给模型。

如果您希望更全面地了解这些概念，该博客中关于按需获取最新知识的检索技术的概述会进行更深入的讲解；此处则侧重于实现细节。

加载 PDF

llm/rag/ingestion/loader.py 使用了 LangChain 的 PyPDFLoader，该工具可将 PDF 的每一页转换为一个 Document 对象。

from pathlib import Path
from langchain_community.document_loaders import PyPDFLoader

PDF_PATH = Path("docs/manual.pdf")

def load_manual():
    loader = PyPDFLoader(
        str(PDF_PATH),
    )
    documents = loader.load()
    return documents

一个 Document 对象包含两部分内容：存储在 page_content 中的提取文本，以及描述该文本来源的 metadata 字典。正是这些元数据使得后续答案能够引用特定页面。从概念上讲，每个加载后的页面结构如下：

Document
├── page_content
│   └── "To create a new demo data..."
│
└── metadata
    ├── source: docs/manual.pdf
    └── page: 12

PyPDFLoader通常会同时记录以零为起点的page索引以及便于人类阅读的page_label。下面的上下文构建器使用的是page_label，它与读者在PDF中看到的页码是一致的。

将页面分割成块

即便只搜索整个文档作为一个整体，其结果也会相当粗糙。llm/rag/ingestion/chunker.py会使用RecursiveCharacterTextSplitter来对文档进行分割。

from langchain_text_splitters import (
    RecursiveCharacterTextSplitter,
)
from langchain_core.documents import Document

def chunk_documents(
    documents: list[Document],
) -> list[Document]:
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=1000,
        chunk_overlap=150,
        separators=[
            "\n\n",
            "\n",
            ". ",
            " ",
            "",
        ],
    )
    return splitter.split_documents(
        documents,
    )

该分割工具的目标是将文本分成约1,000个字符的段落，各段落之间有150个字符的重叠。分割时会按顺序尝试不同的分隔位置：优先在段落边界处分割，其次是行尾、句子末尾和空格处，最后才考虑在单词中间进行分割。设置重叠区域是为了避免某个事实横跨多个分割点，通过在两侧重复少量文本，可以降低相关句子被截断的概率。

这些数值只是参考起点，并非固定规则。实际的分段大小取决于文档的写作方式以及检索所需的精确度，因此应将其视为需要根据具体需求调整的参数。博客中关于保留证据的分段处理的文章进一步探讨了这种权衡关系。

持久化的Chroma集合

llm/rag/ingestion/chroma.py会打开一个PersistentClient，该客户端将数据存储在磁盘上，并返回手册集合；如首次使用则会创建该集合。

import chromadb
from llm.rag.config import (
    CHROMA_PATH,
    MANUAL_COLLECTION_NAME,
)

def get_chroma_client():
    return chromadb.PersistentClient(
        path=CHROMA_PATH,
    )

def get_manual_collection():
    client = get_chroma_client()
    return client.get_or_create_collection(
        name=MANUAL_COLLECTION_NAME,
    )

路径与名称来自llm/rag/config.py，该文件会读取环境变量，若未设置则使用合理的默认值：

import os

CHROMA_PATH = os.getenv(
    "CHROMA_PATH",
    "./chroma_data",
)
MANUAL_PATH = os.getenv(
    "MANUAL_PATH",
    "docs/manual.pdf",
)
MANUAL_COLLECTION_NAME = os.getenv(
    "MANUAL_COLLECTION_NAME",
    "manual",
)

请在.env文件中添加相应的配置项：

CHROMA_PATH=./chroma_data
MANUAL_PATH=docs/manual.pdf
MANUAL_COLLECTION_NAME=manual

当前并未配置任何嵌入模型，这是演示版的刻意设计。当创建集合时未指定嵌入函数，Chroma会使用其内置的默认模型：每次添加文档时，Chroma都会自行计算这些文档的嵌入向量，并将它们与文本及元数据一起存储。该默认模型在本地运行，首次使用时会被下载，因此首次索引时可能会因下载而暂停。

最终会在 chroma_data/ 目录中生成一个本地的、持久化的向量存储。由于该存储完全源自 PDF，因此应将其与 .env 一起添加到 .gitignore 中。

索引任务

llm/rag/ingestion/indexer.py 负责协调所有的数据导入步骤。

from pathlib import Path
from llm.rag.config import MANUAL_PATH
from llm.rag.ingestion.loader import load_manual
from llm.rag.ingestion.chunker import chunk_documents
from llm.rag.ingestion.chroma import get_manual_collection

def index_manual():
    collection = get_manual_collection()
    if collection.count() > 0:
        print(
            f"Manual already indexed "
            f"({collection.count()} chunks)."
        )
        return
    manual_path = Path(
        MANUAL_PATH,
    )
    if not manual_path.exists():
        raise FileNotFoundError(
            f"Manual not found: {manual_path}"
        )
    documents = load_manual()
    print(
        f"Loaded {len(documents)} pages."
    )
    chunks = chunk_documents(
        documents,
    )
    print(
        f"Created {len(chunks)} chunks."
    )
    collection.add(
        ids=[
            f"manual-chunk-{i}"
            for i in range(len(chunks))
        ],
        documents=[
            chunk.page_content
            for chunk in chunks
        ],
        metadatas=[
            chunk.metadata
            for chunk in chunks
        ],
    )
    print(
        f"Stored {len(chunks)} chunks."
    )

让我们详细看看它的功能。它会打开对应的集合，如果其中已存在数据块则立即返回，这样重复运行也不会产生问题。接着它会检查 PDF 是否存在，若不存在则会抛出明确的错误信息。之后它加载页面内容，将其分割成数据块，并通过一次调用将所有内容连同稳定的标识符（如 manual-chunk-0、manual-chunk-1 等）、数据块文本及其元数据一起添加到 Chroma 中。进度提示会显示已处理了多少页面和数据块。

提前返回会导致一个问题：如果你修改了手册并再次运行脚本，由于集合并非为空，因此不会发生任何变化。要应用这些更改，你必须在重新建立索引之前删除该集合（或 chroma_data/ 目录），或者用能够主动插入或重建数据的逻辑替换原有的保护代码。

列表中并未显示 scripts/index_manual.py；它只需导入 index_manual 并调用它即可。从项目根目录将其作为模块运行一次：

python -m scripts.index_manual

该程序会一次性读取PDF文件，将其分割成多个块，嵌入这些块并存储起来。终端输出会显示相关计数信息。在简化的演示版本中，该PDF文件仅包含一页内容，上面写着一条规则：“演示数据只能提供给管理员用户”，这足以验证数据检索功能是否正常。此后，启动FastAPI时便不再操作PDF文件，因为向量数据早已被保存下来。

查询集合

仅建立索引对智能体没有任何帮助；它需要一种搜索方式。llm/rag/retrieval/retriever.py文件封装了Chroma的查询API。

from dataclasses import dataclass
from typing import Any
from llm.rag.ingestion.chroma import (
    get_manual_collection,
)

@dataclass
class RetrievedChunk:
    content: str
    metadata: dict[str, Any]
    distance: float

def retrieve_manual(
    query: str,
    n_results: int = 5,
) -> list[RetrievedChunk]:
    collection = get_manual_collection()
    results = collection.query(
        query_texts=[query],
        n_results=n_results,
        include=[
            "documents",
            "metadatas",
            "distances",
        ],
    )
    documents = results["documents"] or []
    metadatas = results["metadatas"] or []
    distances = results["distances"] or []
    retrieved_chunks = []
    for document, metadata, distance in zip(
        documents[0],
        metadatas[0],
        distances[0],
    ):
        retrieved_chunks.append(
            RetrievedChunk(
                content=document,
                metadata=dict(metadata)
                if metadata else {},
                distance=distance,
            )
        )
    return retrieved_chunks

该函数将问题以query_texts的形式发送，最多请求五个结果，并获取文档、其元数据以及它们之间的距离。Chroma会为每个查询返回一个列表，因此代码会读取每个字段的索引[0]；而or []这种处理方式则可用于防止字段缺失的情况。每个检索到的结果都会被封装为RetrievedChunk数据类，这样其他代码就不必依赖Chroma的响应结构。

由于查询与存储的片段使用的是同一模型，因此匹配是基于语义的。像“我该如何添加新的演示数据？”这样的问题，即使相关内容从未出现“添加新”这类字眼，也能找到关于创建或授予演示数据的段落。距离值可以显示每条匹配结果的相似程度——数值越低表示相似度越高。如果希望排除相似度较低的匹配结果，而非总是将五个片段传递给模型，这个数值日后会非常有用。

有了这一机制，RAG的检索部分就能正常工作。接下来要做的就是将检索结果传递给模型。

将片段转化为上下文

llm/rag/context.py负责将检索到的结果格式化为模型能够读取的字符串。

from llm.rag.retrieval.retriever import (
    RetrievedChunk,
)

def build_context(
    chunks: list[RetrievedChunk],
) -> str:
    context_parts = []
    for chunk in chunks:
        page = chunk.metadata.get(
            "page_label",
        )
        context_parts.append(
            f"Source: User Guide, page {page}\n"
            f"{chunk.content}"
        )
    return "\n\n---\n\n".join(
        context_parts,
    )

每个代码块的前面都有一行源代码，注明用户指南的名称及其页面标签，各个代码块之间则用分隔符隔开。正是这行源代码让模型能够说明答案的来源，同时也为用户提供了验证的途径。

第6步：将手册作为工具呈现

整个流程已经完成：页面被加载并分割成多个代码块，这些代码块存储在ChromaDB中，相关的代码块可以被查找并格式化。不过，代理程序却不知道这一切的存在。这时之前提到的架构设计思路就派上用场了：文档搜索也就变成了另一种工具而已。

llm/tools/manual_tool.py 中定义了 search_user_manual 函数，该函数接收查询内容，检索五个相关片段，并以格式化后的上下文形式返回。如果未找到相关内容，它会返回一条明确消息说明手册中没有涉及该问题，这样模型就有具体的信息可以传递，而非返回空字符串。

from langchain.tools import tool
from llm.rag.context import build_context
from llm.rag.retrieval.retriever import retrieve_manual

@tool
def search_user_manual(
    query: str,
) -> str:
    """
    Search the application user manual.
    Use this tool when the user asks about application
    behavior, instructions, rules, limitations, or
    how something works.
    """
    chunks = retrieve_manual(
        query=query,
        n_results=5,
    )
    if not chunks:
        return (
            "The manual does not contain enough "
            "information to answer this question."
        )
    return build_context(
        chunks,
    )

与之前一样，文档字符串是该工具向模型展示的功能说明。其中指出此工具适用于询问行为、操作指南、规则、限制以及事物运作方式等问题，这与系统提示语的内容是一致的。

现在需要将这两个工具都注册到代理中：

from langchain.agents import create_agent
from llm.provider import model
from llm.prompts import SYSTEM_PROMPT
from llm.tools.demo_tool import (
    get_my_solar_system,
)
from llm.tools.manual_tool import (
    search_user_manual,
)

agent = create_agent(
    model=model,
    tools=[
        get_my_solar_system,
        search_user_manual,
    ],
    system_prompt=SYSTEM_PROMPT,
)

注意该代码片段中的导入语句：它引用了来自整个应用程序的 get_my_solar_system，而演示工具模块定义的是 get_my_demo_data。请在导入语句和 tools 列表中都使用 get_my_demo_data，否则模块将无法导入。

为何基于工具的设计能在应用扩展时依然有效

智能体无需一个包含应用程序所有信息的庞大提示词，而是拥有较为有限且可控的功能。要为助手添加新功能，只需编写新的工具并对其进行注册即可；HTTP 层不会因此改变。演示版本仅保留一个数据工具和一个文档工具，这样结构一目了然，但在完整的应用程序中，同样的架构可以支持更多工具。

第7步：将已认证的用户传递给智能体

演示工具仍然返回硬编码的数据。真正的工具必须能够识别请求者，而该应用已经做到了这一点：FastAPI在身份验证依赖项中解析了用户信息。缺失的部分是将该用户信息传递到代理运行过程中。LangChain将此称为运行时上下文。

在llm/context.py中定义上下文的结构：

from dataclasses import dataclass
from auth.dependencies import MockUser

@dataclass
class AgentContext:
    user: MockUser

当调用代理时就会传递这个对象，而且这种区分非常重要。用户信息属于请求范围内的数据，它归属于当前的HTTP请求而非整个对话过程，绝不能以模型可以读取或修改的消息形式存储。将其排除在消息历史之外还能防止提示语诱使代理假扮成其他用户。在完整的应用中，同一个上下文对象还会包含数据库会话以及正在编辑的项目ID等信息。

演示代码仅在定义类时停止，因此还有两个连接需要您自行建立，建议查阅最新的LangChain文档以获取准确的API信息。首先，在构建智能体时声明架构，通常是通过向create_agent函数传递context_schema=AgentContext参数来实现。其次，让工具读取该架构：在LangChain 1.x版本中，工具可以接受一个运行时参数（例如标记为ToolRuntime[AgentContext]），并从其context属性中读取用户信息，而模型无法看到工具的这些参数。真正的get_my_demo_data函数就会在这一点上根据user.id查找记录。

第8步：用于流式处理的编排层

不要在路由器内部调用代理，而应将交互逻辑放在 llm/orchestrator.py 中。这样路由器就能专注于处理 HTTP 请求，而协调器则负责控制消息如何转化为代理的执行。

from collections.abc import Iterator
from langchain_core.messages import (
    AIMessage,
    AIMessageChunk,
    BaseMessage,
    ToolMessage,
)
from langchain_core.runnables import RunnableConfig
from llm.agent import agent
from llm.context import AgentContext

def chat_stream(
    user_message: str,
    user,
) -> Iterator[str]:
    config: RunnableConfig = {
        "configurable": {
            "thread_id": f"user:{user.id}",
        }
    }
    context = AgentContext(
        user=user,
    )
    for chunk, metadata in agent.stream(
        {
            "messages": [
                {
                    "role": "user",
                    "content": user_message,
                }
            ]
        },
        config=config,
        context=context,
        stream_mode="messages",
    ):
        if not isinstance(
            chunk,
            BaseMessage,
        ):
            continue
        if isinstance(
            chunk,
            ToolMessage,
        ):
            continue
        if not isinstance(
            chunk,
            (
                AIMessage,
                AIMessageChunk,
            ),
        ):
            continue
        if isinstance(
            chunk.content,
            str,
        ):
            yield chunk.content

这个函数内容较多，建议逐部分处理。

线程 ID 用于选择对话

第一个代码块负责构建执行配置：

config = {
    "configurable": {
        "thread_id": f"user:{user.id}",
    }
}

LangGraph的检查点器以thread_id为键来存储对话状态。使用相同线程ID的每次运行都会延续同一场对话，其消息日后仍可被读取。这里的线程ID是由用户ID派生而来的，这意味着每个用户只有一场对话。这对于演示来说没问题；而在实际应用中，则需要生成合理的对话ID，允许每个用户拥有多场对话，并在每次请求时验证调用者确实拥有正在访问的线程。

该说明假设存在检查指针，但所有的智能体代码片段均未通过验证。如果没有检查指针，thread_id 将毫无作用，且请求之间也不会保留任何信息。请在 llm/agent.py 中创建一个 InMemorySaver 实例，并通过其 checkpointer 参数将其传递给 create_agent，这样智能体及后续的历史记录功能就能导入同一个实例。

运行时上下文会随程序一同传递

接下来，调度器会将用户封装在上下文对象中：

context = AgentContext(
    user=user,
)

该对象会作为 context= 参数传递给 agent.stream()。这就是在 FastAPI 中建立的标识信息传递给智能体，进而传递给各工具的路径。

过滤数据流

agent.stream()以stream_mode="messages"的参数被调用，当模型生成标记时，它会返回消息片段与元数据的对。并非该流中的所有内容都会呈现给用户。循环会跳过非LangChain消息的内容、ToolMessage对象（如检索到的原始文本等工具输出），仅保留AI消息及其片段，且当内容为纯字符串时才会将其输出。某些服务提供者以部分列表的形式传递的内容会在最后检查中被 silently 跳过，因此如果更换模型后看到空响应，可在此处查找原因。

第9步：返回流式响应

将 chat_stream() 设计为生成器是有意为之。在发送一个字节之前等待完整答案会让用户一直盯着加载指示器，而大型语言模型的回复可能需要数秒时间。通过流式传输，几乎可以立即显示前几句话，这让助手显得响应速度更快。

FastAPI 的 StreamingResponse 可以直接接收生成器。请修改 routers/chat.py 文件：

from fastapi import APIRouter, Depends
from fastapi.responses import StreamingResponse
from auth.dependencies import (
    MockUser,
    get_current_user,
)
from llm.orchestrator import chat_stream

chat_router = APIRouter(
    prefix="/chat",
    tags=["Chat"],
)

@chat_router.post("")
def ask_ai(
    message: str,
    current_user: MockUser = Depends(
        get_current_user,
    ),
):
    return StreamingResponse(
        chat_stream(
            user_message=message,
            user=current_user,
        ),
        media_type="text/plain",
    )

响应以 text/plain 格式发送，每生成一段内容就会立即写入连接中。由于 chat_stream 是普通的（同步的）生成器，Starlette 会在工作线程中遍历它，因此不会阻塞事件循环。如果日后需要在客户端获取结构化事件（例如在工具运行时显示“正在查找手册……”），Server-Sent Events 是一个自然的下一步选择。

完整的请求路径如下：客户端向 /chat 发送请求，FastAPI 对调用者进行身份验证，chat_stream() 启动代理进程，模型决定是否调用工具，相应的工具执行并返回结果，模型将答案写入，最后这些数据以令牌形式传回客户端。

关键点在于 FastAPI 本身从不直接运行模型。路由器负责处理 HTTP 请求，协调器驱动代理进程，代理决定需要哪些信息，而工具则负责获取数据。各层之间可以独立修改而不影响其他部分。

第10步：读取对话历史记录

由于代理进程会保存检查点，因此每次交互后其状态都会被存储下来。这便能够向返回的用户展示他们之前的对话内容。有两个辅助函数负责完成这项工作。

读取原始检查点

第一个函数会加载线程的最新检查点，并返回messages通道；如果该线程从未被使用过，则返回空列表：

def get_conversation_messages(
    thread_id: str,
) -> list[BaseMessage]:
config: RunnableConfig = {
        "configurable": {
            "thread_id": thread_id,
        }
    }
    checkpoint = checkpointer.get(
        config,
    )
    if checkpoint is None:
        return []
    return checkpoint[
        "channel_values"
    ].get(
        "messages",
        [],
    )

如果复制此代码，需修正config赋值的缩进：它必须位于函数体内部，否则Python会抛出错误。此外还需导入BaseMessage、RunnableConfig以及共享的checkpointer实例。

函数返回的是代理的原始状态，其中包含的内容远不止用户所记得的聊天记录。当代理调用工具时，LangGraph会记录一条包含工具调用的AI消息，以及另一条包含结果的独立工具消息。这些都是实现细节，前端无需对其进行解读。

格式化用于显示的消息

第二个函数用于构建面向用户的界面：

def get_conversation_messages_for_display(
    thread_id: str,
) -> list[dict[str, str]]:
    display = []
    for message in get_conversation_messages(
        thread_id,
    ):
        if isinstance(
            message,
            HumanMessage,
        ):
            content = _extract_text_content(
                message.content,
            )
            if content.strip():
                display.append(
                    {
                        "type": "human",
                        "content": content,
                    }
                )
            continue
        if isinstance(
            message,
            AIMessage,
        ):
            content = _extract_text_content(
                message.content,
            )
            if content.strip():
                display.append(
                    {
                        "type": "ai",
                        "content": content,
                    }
                )
    return display

它仅保留HumanMessage和AIMessage对象，提取它们的文本，丢弃空对象，然后返回包含type和content键的简单字典。工具消息不会显示，因为它们不属于这两种允许的类型。空的人工智能消息也会被丢弃，这一点很重要，因为仅请求工具调用的人工智能消息通常没有文本内容。

该功能依赖于一个未展示的辅助函数_extract_text_content。它的作用是：当内容为字符串时直接返回；当内容为多个文本部分的列表时，则将这些部分拼接在一起。此外还需要导入HumanMessage和AIMessage。

历史记录接口

通过 routers/chat.py 中的 GET 路由来展示显示视图。该路由会获取与聊天端点相同的线程 ID，并将其与消息一起返回。

@chat_router.get("/current")
def get_current_conversation(
    current_user: MockUser = Depends(
        get_current_user,
    ),
):
    thread_id = (
        f"user:{current_user.id}"
    )
    messages = (
        get_conversation_messages_for_display(
            thread_id,
        )
    )
    return {
        "thread_id": thread_id,
        "messages": messages,
    }

页面加载时，聊天界面可以调用此接口，在用户输入任何内容之前就渲染出现有的对话内容：

GET /chat/current

响应内容如下：

{
    "thread_id": "user:user-123",
    "messages": [
        {
            "type": "human",
            "content": "How much demo data do I have?"
        },
        {
            "type": "ai",
            "content": "You currently have 18 demo data."
        }
    ]
}

为何不直接返回原始状态

智能体的内部状态与用户看到的对话内容是两回事。随着功能的增加，状态中会累积工具调用记录、工具返回结果、中间处理步骤、模型元数据以及其他相关信息。如果全部返回，将会使前端与智能体的内部结构紧密耦合，还可能泄露本不应显示的工具输出内容。后端应当明确界定公开可用的对话历史范围，仅返回该范围内的内容。

单个请求的端到端处理流程

在所有组件都已就位的情况下，有一点需要明确：该模型永远不会直接访问您的数据库或PDF文件。它只能请求调用某个工具。该工具以普通Python代码的形式运行，并具有常规的访问控制权限，执行相应操作后返回文本，模型再利用这些文本来生成答案。正是这一界限确保了该助手能够安全地嵌入到处理真实数据的应用程序中。

在Swagger UI中测试

FastAPI会自动生成交互式文档，因此在测试时无需额外的客户端。首先启动服务器：

python -m uvicorn main:app --reload

接着打开在/docs路径下提供的交互式API文档，设置X-Demo-User请求头，然后尝试三种交互操作：

数据相关问题，例如询问用户拥有哪些演示数据。智能体应判断需要使用演示工具，调用该工具，并根据返回的项目详情给出答案。在完整应用中，同类工具会查询用户的实际记录。
文档相关问题，例如询问谁可以接收演示数据。智能体应进行手动搜索，找到对应的索引规则，并回答只有管理员用户可以接收。
历史记录接口，即GET /chat/current，该接口应返回前两个问题中人类与AI的对话内容，而不包含任何工具相关的消息。

如果前两个问题能基于工具输出给出答案，而第三个问题能显示清晰的对话记录，那就说明各层功能都正常运行。

在投入生产之前

该演示有意简化了若干组件。在真实用户开始依赖该服务之前，需要重新审视这些部分。

持久对话状态

InMemorySaver非常适合开发使用，但当进程重启时它所存储的所有内容都会消失，而且无法在负载均衡器背后的多个API实例之间共享。应使用由数据库或其他持久存储支持的检查点机制，以确保对话状态在部署后依然存在，并且所有实例都能看到相同的状态。关于InMemorySaver实际存储的内容及其存储方式，可参阅博客中关于InMemorySaver如何组织检查点、写入数据及处理二进制文件的详细说明。

真正的向量存储部署

使用本地的 Chroma 目录来演示流程是可行的，但它并非生产环境用的基础设施。应将 Chroma 作为持久化服务运行，或迁移到适合您技术栈的托管向量数据库。无论选择哪种方式，都必须确保其具有持久性、能被备份，并且所有应用实例都能访问。

索引操作不在 API 范围内

该演示已经做出了一个正确的决策：索引是通过单独的脚本来执行的，而 API 仅负责数据检索。

python -m scripts.index_manual

服务器在启动时不会加载 PDF、将其分割成块或计算嵌入向量。索引操作是离线进行的数据导入，而检索则是处理请求的一部分。将这两者分开意味着 API 不需要检测文档变更或重新构建任何内容。

在实际应用中，可进一步将相同的索引代码作为独立的导入任务来运行，该任务可通过部署流水线定时触发，或在有新文档上传时由工作进程自动启动。架构保持不变，只是任务变得自动化、可重复且能够独立部署。这样一来，各部分的职责便能清晰划分：

导入任务：加载文档、将其分割成块、嵌入向量数据，并更新向量存储。
FastAPI服务：接收查询请求、检索相关文档块并生成响应。
向量存储：持久化查询时使用的索引后的向量表示。

在实现自动化时，请记住索引器中的提前返回机制；如果任务默默跳过重新索引操作，那比不执行任务还要糟糕。

显式的嵌入模型

使用 Chroma 的默认嵌入函数可以让演示无需额外配置，但生产系统应明确选择并配置自己的嵌入模型。这样既能保证结果的可重复性，又能让开发者掌控质量、成本、延迟以及嵌入计算的地点。有一条规则是不可商量的：索引和查询必须使用相同的嵌入模型。一旦更改，就需要重新对所有数据建立索引。

可观测性与错误处理

当智能体开始运行后，了解它的操作情况与确保其正常工作同样重要。在得到最终答案之前，一个请求可能涉及多次模型调用、一次或多次工具调用以及检索步骤。仅记录最终答案在出现问题时几乎无法提供任何信息。需要对整个流程进行监控：

工具调用：使用了哪些工具，参数是什么，每项调用耗时多久。

模型延迟：每次LLM请求所耗用的时间。

令牌使用量与成本：当一条用户消息可能触发多次模型调用时，这一指标非常重要。

工具故障：工具应返回受控的错误信息，而非导致请求崩溃。

服务提供方故障：需妥善处理速率限制、超时以及模型不可用等问题，理想情况下还应具备备用模型。

运行追踪：记录每次运行中模型调用、工具调用及响应的顺序。

检索质量：如果手动搜索持续返回无关内容，其原因通常出在分块处理、嵌入向量、查询语句或检索设置上，而非LLM本身。

目标在于让智能体永远不是黑箱。对于任何一次运行，都应能够明确它做了什么、调用了哪些工具、每一步耗时多久以及在哪里出错。具体的工具选择取决于你所使用的技术栈，但这一原则是不变的。

关键要点

无需大型人工智能平台即可在功能复杂的应用中添加实用助手。只需从能够解决实际问题的最小组件集开始即可。
将文档检索视作众多工具之一。这样智能体就能以统一的方式获取产品知识与用户数据。
将身份信息保存在运行时上下文中，而非消息中。用户属于请求的一部分，工具应从那里读取相关信息。
将HTTP、编排机制、智能体与工具分开。每层都保持简洁，要增加新功能只需添加相应的工具即可。

通过检查点机制，你几乎可以免费获取历史记录，但得到的是经过筛选的视图，而非原始的智能体状态。

流式输出响应、离线建立索引、固定嵌入模型，并在真实用户到来之前对每一步进行监控。