本文详细介绍了如何搭建一个带监控面板的RAG项目，从环境配置、项目骨架搭建到文档加载、AI问答引擎的构建，以及CLI和API双入口的实现。此外，还介绍了如何使用LangSmith和Phoenix进行项目监控，并提供常见报错速查表和项目定制化指南，帮助读者全面掌握RAG项目的搭建和优化。

---

搭建一个带监控面板 RAG 项目

---

目录

1. 1. 这到底是什么（用大白话讲）
2. 2. 你需要准备什么
3. 3. 第一步：搭好项目骨架
4. 4. 第二步：配置中心 config.py
5. 5. 第三步：让数据库自动建好 store.py
6. 6. 第四步：把文档"翻译"成 AI 能懂的格式 loader.py
7. 7. 第五步：让 AI 学会回答你的问题 engine.py
8. 8. 第六步：CLI 和 API 双入口
9. 9. 第七步：第一次运行！
10. 10. 第八步：加个监控面板（LangSmith）
11. 11. 第九步：LangSmith Studio 本地 Chat UI
12. 12. 第十步：Phoenix 零上报本地监控
13. 13. 常见报错速查表
14. 14. 换成你自己的文档 / 模型 / 数据库

---

1. 这到底是什么

---

一个比喻

想象你有一本 500 页的教科书。考试时不允许翻书，但你可以在考前做一件事：把书里每个知识点写成小卡片，分类编号，放进卡片盒。

考试时看到题目，你从卡片盒里找出最相关的 5 张卡片，铺在桌上，然后开始答题。

RAG 就是这个过程，只不过：

• • 教科书 = 你的文档文件夹
• • 小卡片 = 文档切成的"数据块"（chunk）
• • 卡片盒 = 向量数据库（我们用的 pgvector）
• • 你 = 大语言模型（我们用的通义千问）

为什么不用 ChatGPT 直接上传文件？

上传文件有大小限制、每次对话都要重新上传、不能批量管理几十个文件。RAG 一次性把文件"消化"好，以后随时问，回答还带出处。

你最终得到什么

# 命令行里直接问$ python -m rag.run q "红绿灯模型是什么？"
```![](http://cdn.zhipoai.cn/06b61976.jpg)

```plaintext
# 或者启动 API，其他程序也能调用$ python -m rag.run api# → http://localhost:8000/query

---

2. 你需要准备什么

---

硬件 / 软件检查

打开终端（PowerShell 或 CMD），逐条运行确认：

# 1. Python 版本 >= 3.11python --version# 期望输出: Python 3.11.x 或更高# 2. PostgreSQL 是否运行中psql -U postgres -c "SELECT 1"# 如果报错 "role 'postgres' does not exist" 或连不上，先解决数据库问题# 3. pgvector 扩展是否可用psql -U postgres -c "CREATE EXTENSION IF NOT EXISTS vector"# 如果报错 "extension 'vector' is not available"，需要先安装 pgvector

申请两个 API Key

服务	用途	去哪里申请	费用
阿里云 DashScope	文本嵌入 + 大模型回答	dashscope.aliyun.com	新用户有免费额度
LangSmith	可选，监控面板	smith.langchain.com	个人用免费

不需要 GPU，不需要显卡。 计算全部走云端 API，你的电脑只负责"传文件、收答案"。

---

3. 第一步：搭好项目骨架

---

3.1 创建文件夹

mkdir D:\qcgs\rag

如果你的项目在其他目录，把 D:\qcgs 换成你的路径。后面所有路径同理。

3.2 创建依赖清单

在 D:\qcgs\rag\pyproject.toml 写入：

[project]name = "qcgs-rag"version = "0.1.0"requires-python = ">=3.11"dependencies = [    "llama-index-core>=0.12.0",    "llama-index-vector-stores-postgres>=0.8.0",    "llama-index-embeddings-dashscope>=0.5.0",    "llama-index-llms-dashscope>=0.6.0",    "langsmith>=0.3.0",    "langgraph>=0.4.0",    "langchain-core>=0.3.0",    "langgraph-cli[inmem]>=0.1.0",    "psycopg2-binary>=2.9.0",    "fastapi>=0.115.0",    "uvicorn[standard]>=0.34.0",    "click>=8.0.0",    "python-dotenv>=1.0.0",    "arize-phoenix>=16.0.0",    "openinference-instrumentation-llama-index>=4.0.0",]

3.3 创建配置文件

在 D:\qcgs\rag\.env 写入（把你自己的 Key 填进去）：

# --- 必填 ---DASHSCOPE_API_KEY=sk-你的阿里云key# --- 数据库（用默认值即可）---PG_HOST=localhostPG_PORT=5432PG_USER=postgresPG_PASSWORD=你的数据库密码PG_DATABASE=qcgs_rag# --- LangSmith（可以先不开，后面再加）---LANGSMITH_API_KEY=你的langsmith-keyLANGSMITH_PROJECT=qcgs-ragLANGSMITH_TRACING=false# --- Phoenix 本地监控（可以先不开，后面再加）---PHOENIX_ENABLED=false

3.4 创建 .gitignore

在 D:\qcgs\rag\.gitignore 写入：

.env.venv/__pycache__/*.pyc

为什么重要？ .env 里有你的 API Key，绝对不能提交到 Git。.gitignore 就是告诉 Git “这些文件别管”。

3.5 创建包标记文件

在 D:\qcgs\rag\__init__.py 写入：

# 比所有 import 更早执行，确保 .env 中的环境变量已加载# （LangSmith 和 LlamaIndex 需要在导入前看到这些变量）from dotenv import load_dotenvfrom pathlib import Pathload_dotenv(Path(__file__).parent / ".env")# 如果 Phoenix 开启，在 LlamaIndex 被使用前完成 OpenTelemetry 注册from .config import setup_phoenix as _setup_phoenix_setup_phoenix()

解释：Python 导入一个包时，会自动先执行 __init__.py。我们把 .env 加载放在这里，保证任何后续代码读到环境变量时，Key 已经就位了。

3.6 安装依赖

cd D:\qcgs\ragpip install -e .

或者用 uv（更快）：

cd D:\qcgs\raguv pip install llama-index-core llama-index-vector-stores-postgres llama-index-embeddings-dashscope llama-index-llms-dashscope langsmith langgraph langchain-core "langgraph-cli[inmem]" psycopg2-binary fastapi "uvicorn[standard]" click python-dotenv arize-phoenix openinference-instrumentation-llama-index

验证：

python -c "import llama_index.core; print('OK')"# 期望输出: OK

---

4. 第二步：配置中心

---

这个文件的作用：把所有可变参数（Key、数据库密码、模型名）集中在一个地方管理。以后要改什么只改这一个文件，不用满世界找。

在 D:\qcgs\rag\config.py 写入：

"""所有配置集中管理。"""import osfrom pathlib import Path# .env 已在 __init__.py 中加载，这里直接读 os.getenv 即可# --- 文档路径 ---DOCS_DIR = Path(os.getenv("DOCS_DIR", str(Path(__file__).parent.parent)))RAG_DIR = Path(__file__).parent# --- PostgreSQL ---PG_HOST = os.getenv("PG_HOST", "localhost")PG_PORT = int(os.getenv("PG_PORT", "5432"))PG_USER = os.getenv("PG_USER", "postgres")PG_PASSWORD = os.getenv("PG_PASSWORD", "postgres")PG_DATABASE = os.getenv("PG_DATABASE", "qcgs_rag")# --- 阿里云 DashScope ---DASHSCOPE_API_KEY = os.getenv("DASHSCOPE_API_KEY", "")EMBEDDING_MODEL = "text-embedding-v3"   # 把文本转成向量的模型EMBEDDING_DIM = 1024                    # 这个模型输出的向量维度LLM_MODEL = "qwen-plus"                 # 回答问题的大模型# --- 分块参数 ---SENTENCE_WINDOW_SIZE = 3    # 检索到一句话时，前后各多拿几句给 AI 做上下文SENTENCE_CHUNK_SIZE = 2048# 每块最多多少 tokenSENTENCE_CHUNK_OVERLAP = 128# 块与块之间重叠多少，避免一句被切断# --- 检索参数 ---TOP_K = 5                    # 每次检索返回最相关的几块SIMILARITY_CUTOFF = 0.3      # 相关度低于此值的丢弃# --- LangSmith 监控（可选） ---LANGSMITH_API_KEY = os.getenv("LANGSMITH_API_KEY", "")LANGSMITH_PROJECT = os.getenv("LANGSMITH_PROJECT", "qcgs-rag")LANGSMITH_TRACING = os.getenv("LANGSMITH_TRACING", "false").lower() == "true"_LANGSMITH_INITIALIZED = False# --- API 服务 ---API_HOST = os.getenv("API_HOST", "0.0.0.0")API_PORT = int(os.getenv("API_PORT", "8000"))# --- Phoenix 本地监控（完全离线，零上报）---PHOENIX_ENABLED = os.getenv("PHOENIX_ENABLED", "false").lower() == "true"PHOENIX_ENDPOINT = os.getenv("PHOENIX_ENDPOINT", "http://localhost:6006/v1/traces")def _validate_config():    """启动时检查必填项，没配好就直接报错，不给后面留坑。"""    missing = []    if not DASHSCOPE_API_KEY:        missing.append("DASHSCOPE_API_KEY")    if missing:        raise RuntimeError(            f"缺少必要的环境变量: {', '.join(missing)}。"            f"请在 {Path(__file__).parent / '.env'} 中填写。"        )_validate_config()def get_tracer():    """获取 LangSmith 追踪器。如果没开追踪，返回一个什么都不做的假追踪器。"""    global _LANGSMITH_INITIALIZED    if not LANGSMITH_TRACING:        class _NoOpTracer:            def __call__(self, func=None, *, name=None, run_type=None):                return func if func is not None else (lambda f: f)            def __enter__(self): return self            def __exit__(self, *args): pass        return _NoOpTracer()    if not _LANGSMITH_INITIALIZED:        _LANGSMITH_INITIALIZED = True        os.environ["LANGSMITH_TRACING_V2"] = "true"        os.environ.setdefault("LANGSMITH_ENDPOINT", "https://api.smith.langchain.com")        print(f"[LangSmith] 追踪已开启 — 项目: {LANGSMITH_PROJECT}")    from langsmith import traceable    return traceabledef setup_phoenix():    """启动 Phoenix 本地追踪（仅在 PHOENIX_ENABLED=true 时生效）。    必须在任何 LlamaIndex 操作前调用，因为 OpenInference    需要在 LlamaIndex 内部组件初始化前完成猴子补丁。    """    if not PHOENIX_ENABLED:        return    from phoenix.otel import register    from openinference.instrumentation.llama_index import LlamaIndexInstrumentor    _ = register(project_name="qcgs-rag", endpoint=PHOENIX_ENDPOINT, batch=True)    LlamaIndexInstrumentor().instrument()    print(f"[Phoenix] 本地追踪已开启 — UI: http://localhost:6006")

关键设计：

• • _validate_config() — 如果忘填 Key，import 时直接报错，不会跑到一半才发现
• • get_tracer() — 开了 LangSmith 就追踪，没开就零开销，不需要改代码
• • setup_phoenix() — 开关控制，Phoenix 未安装或未开启时完全不影响正常运行

验证：

cd D:\qcgs && python -c "from rag.config import DOCS_DIR; print(f'文档目录: {DOCS_DIR}')"# 期望输出: 文档目录: D:\qcgs （或你的项目路径）

---

5. 第三步：让数据库自动建好

---

这个文件的作用：第一次运行时自动创建数据库、开启向量扩展、建好表。你不需要手动敲一行 SQL。

在 D:\qcgs\rag\store.py 写入：

"""自动管理 pgvector 数据库：建库、扩展、索引配置。"""import loggingimport psycopg2from psycopg2 import sqlfrom psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMITfrom llama_index.vector_stores.postgres import PGVectorStorefrom .config import (    PG_HOST, PG_PORT, PG_USER, PG_PASSWORD, PG_DATABASE,    EMBEDDING_DIM,)logger = logging.getLogger(__name__)def ensure_database():    """如果 qcgs_rag 库不存在，就创建它。"""    with psycopg2.connect(        host=PG_HOST, port=PG_PORT, user=PG_USER,        password=PG_PASSWORD, dbname="postgres",    ) as conn:        conn.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)        with conn.cursor() as cur:            cur.execute(                "SELECT 1 FROM pg_database WHERE datname = %s",                (PG_DATABASE,),            )            if cur.fetchone() is None:                cur.execute(                    sql.SQL("CREATE DATABASE {}").format(                        sql.Identifier(PG_DATABASE)                    )                )def ensure_pgvector_extension():    """开启 pgvector 扩展（PostgreSQL 的向量计算插件）。"""    with psycopg2.connect(        host=PG_HOST, port=PG_PORT, user=PG_USER,        password=PG_PASSWORD, dbname=PG_DATABASE,    ) as conn:        conn.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)        with conn.cursor() as cur:            cur.execute("CREATE EXTENSION IF NOT EXISTS vector")def drop_table():    """重建索引时清空旧数据。"""    with psycopg2.connect(        host=PG_HOST, port=PG_PORT, user=PG_USER,        password=PG_PASSWORD, dbname=PG_DATABASE,    ) as conn:        conn.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)        with conn.cursor() as cur:            cur.execute("DROP TABLE IF EXISTS data_llamaindex CASCADE")def get_vector_store():    """返回一个配置好的向量数据库连接。    使用 HNSW 索引：检索速度快、精度高，适合十万级以内的文档。    """    return PGVectorStore.from_params(        database=PG_DATABASE,        host=PG_HOST,        port=str(PG_PORT),        user=PG_USER,        password=PG_PASSWORD,        table_name="llamaindex",        embed_dim=EMBEDDING_DIM,        hnsw_kwargs={            "m": 16,            "ef_construction": 200,            # ⚠️ 下面三个必须同时带上 hnsw_ 前缀，缺一个查询就报错            "hnsw_ef_search": 200,            "hnsw_m": 16,            "hnsw_ef_construction": 200,        },    )def setup_store():    """一键完成全部数据库准备工作。"""    ensure_database()    ensure_pgvector_extension()    return get_vector_store()

需要理解的两个概念：

概念	一句话解释
with 语句	自动关闭数据库连接。如果不用 with，报错时连接就泄露了，积少成多数据库会挂
ISOLATION_LEVEL_AUTOCOMMIT	建库、建扩展这类操作必须在"自动提交"模式下执行，不能放在事务里

验证：

cd D:\qcgs && python -c "from rag.store import setup_store; setup_store(); print('数据库就绪')"

---

6. 第四步：文档"翻译"

---

这个文件的作用：把原始 markdown 文件切成大小合适的块，调用阿里云 API 把每块转成向量（一串数字）。这是 RAG 最关键的一步。

什么是"向量"？（30 秒理解）

"今天天气真好" → embedding API → [0.12, -0.45, 0.78, ..., 0.33]                                       ↑                                  一串 1024 个数字含义相近的句子，它们的向量在数学空间里距离很近。"今天天气真好" 和 "阳光明媚" → 向量距离近 → 检索时会被一起找到"今天天气真好" 和 "数据库索引" → 向量距离远 → 检索时不会被混在一起

什么是"父子分块"？（核心技巧）

原始文档很长的段落        ↓ 切成小块（子块）  子块1  子块2  子块3  子块4  ← 这些用于向量检索（精确匹配）        ↓ 检索时自动替换  "子块2 + 前后各 N 句" ← 这个更大的窗口给 AI 看（上下文完整）

效果：找得准（小块）+ 理解对（大窗口），不会断章取义。

在 D:\qcgs\rag\loader.py 写入：

"""文档加载 + 智能分块。"""from pathlib import Pathfrom typing import Callablefrom llama_index.core import SimpleDirectoryReaderfrom llama_index.core.node_parser import SentenceWindowNodeParser, SentenceSplitterfrom llama_index.core.ingestion import IngestionPipelinefrom llama_index.embeddings.dashscope import DashScopeEmbeddingfrom llama_index.core.schema import BaseNode, Documentfrom .config import (    DOCS_DIR, RAG_DIR, DASHSCOPE_API_KEY,    SENTENCE_WINDOW_SIZE, SENTENCE_CHUNK_SIZE, SENTENCE_CHUNK_OVERLAP,)def load_documents(docs_dir=None):    """读取目录下所有 .md 文件，跳过代码目录和文档目录。"""    docs_dir = Path(docs_dir) if docs_dir else DOCS_DIR    reader = SimpleDirectoryReader(        input_dir=str(docs_dir),        required_exts=[".md"],        recursive=True,        exclude=[            str(RAG_DIR),             # 别索引自己的代码            str(docs_dir / "docs"),   # 别索引文档目录            str(docs_dir / ".cc"),# 别索引配置目录        ],    )    return reader.load_data()def create_embedding_model():    """创建阿里云的 embedding 模型，把文字变成向量。"""    return DashScopeEmbedding(        model_name="text-embedding-v3",        api_key=DASHSCOPE_API_KEY,        embed_batch_size=10,  # ⚠️ 阿里云限制每次最多 10 条，超过就报错    )def _create_chunked_splitter():    """创建一个分句器，确保每块不超过 2048 token。    为什么需要？阿里云 embedding API 单次最长接受 8192 token。    不加限制的话，markdown 代码块可能被当成一句话，超过 8192 就挂了。    """    _splitter = SentenceSplitter.from_defaults(        chunk_size=SENTENCE_CHUNK_SIZE,        chunk_overlap=SENTENCE_CHUNK_OVERLAP,    )    def _split(text):        return [            node.text            for node in _splitter.get_nodes_from_documents(                [Document(text=text)], show_progress=False            )        ]    return _splitdef create_node_parser():    """创建父子分块器。    子块（3 句一段）→ 用于向量检索，确保精准匹配    父窗口（子块 + 前后各 3 句）→ 存入 metadata，检索后替换，确保 AI 看到完整上下文    """    return SentenceWindowNodeParser.from_defaults(        sentence_splitter=_create_chunked_splitter(),        window_size=SENTENCE_WINDOW_SIZE,        window_metadata_key="window",        original_text_metadata_key="original_text",    )def build_ingestion_pipeline(embed_model=None):    """组装流水线：分块 → embedding。"""    if embed_model is None:        embed_model = create_embedding_model()    node_parser = create_node_parser()    return IngestionPipeline(transformations=[node_parser, embed_model])def ingest_documents(documents=None, pipeline=None, verbose=True):    """执行流水线：把文档列表变成带向量的数据块列表。"""    if documents is None:        documents = load_documents()    if pipeline is None:        pipeline = build_ingestion_pipeline()    if verbose:        print(f"加载了 {len(documents)} 个文档...")        for doc in documents:            print(f"  - {doc.metadata.get('file_name', '?')}")    nodes = pipeline.run(documents=documents, show_progress=verbose)    if verbose:        print(f"生成了 {len(nodes)} 个数据块（每个块都已完成向量化）。")    return nodesdef build_nodes(docs_dir=None, verbose=True):    """一步完成：读文件 → 分块 → 向量化。"""    documents = load_documents(docs_dir)    pipeline = build_ingestion_pipeline()    return ingest_documents(documents, pipeline, verbose)

验证（这一步会调用阿里云 API，可能花几十秒）：

cd D:\qcgs && python -c "from rag.loader import load_documents; docs = load_documents(); print(f'找到 {len(docs)} 个文档')"

---

7. 第五步：让 AI 回答问题

---

这个文件的作用：把前面所有零件串起来——读文档、向量化、存数据库、检索、调用大模型生成回答。

在 D:\qcgs\rag\engine.py 写入：

"""RAG 核心引擎：建索引 + 查询。"""from typing import Anyfrom llama_index.core import VectorStoreIndex, StorageContext, Settingsfrom llama_index.core.postprocessor import MetadataReplacementPostProcessorfrom llama_index.embeddings.dashscope import DashScopeEmbeddingfrom llama_index.llms.dashscope import DashScopefrom .config import (    DASHSCOPE_API_KEY, TOP_K, LLM_MODEL, get_tracer,)from .store import get_vector_store, setup_store, drop_tablefrom .loader import build_nodes# LangSmith 追踪器（如果 LANGSMITH_TRACING=false，这就是个什么都不做的假追踪器）_tracer = get_tracer()def create_llm():    """创建通义千问大模型，负责根据检索结果生成回答。"""    return DashScope(        model_name=LLM_MODEL,        api_key=DASHSCOPE_API_KEY,    )def _make_embed_model():    """创建 embedding 模型。"""    return DashScopeEmbedding(        model_name="text-embedding-v3",        api_key=DASHSCOPE_API_KEY,        embed_batch_size=10,    )def build_index(docs_dir=None, rebuild=True, verbose=True):    """建立向量索引。    流程：    1. 清空旧表    2. 准备好数据库    3. 读取文档 → 分块 → 向量化    4. 把向量存入 pgvector    5. 返回一个可以查询的索引对象    """    embed_model = _make_embed_model()    Settings.embed_model = embed_model    Settings.llm = create_llm()    if rebuild:        if verbose:            print("清空旧索引...")        drop_table()    vector_store = setup_store()    if verbose:        print("正在读取文档并分块...")    nodes = build_nodes(docs_dir, verbose=verbose)    if verbose:        print(f"正在将 {len(nodes)} 个向量写入数据库...")    storage_context = StorageContext.from_defaults(vector_store=vector_store)    index = VectorStoreIndex(nodes, storage_context=storage_context, embed_model=embed_model)    if verbose:        print("索引建立完成！")    return indexdef create_query_engine(index=None, top_k=TOP_K):    """创建查询引擎。    关键：MetadataReplacementPostProcessor    检索时用小块（精准匹配），然后自动替换成大窗口（上下文完整）。    """    if index is None:        vector_store = get_vector_store()        embed_model = _make_embed_model()        Settings.embed_model = embed_model        Settings.llm = create_llm()        index = VectorStoreIndex.from_vector_store(vector_store, embed_model=embed_model)    postprocessor = MetadataReplacementPostProcessor(        target_metadata_key="window",  # 和 loader 里定义的名字对应    )    return index.as_query_engine(        similarity_top_k=top_k,        node_postprocessors=[postprocessor],        response_mode="compact",    )@_tracer(name="qcgs-rag-query", run_type="chain")def query(question, top_k=TOP_K, verbose=False):    """问一个问题，返回 AI 回答 + 引用来源。    Args:        question: 你的问题        top_k: 检索几条最相关的内容给 AI 参考        verbose: 要不要打印检索详情    Returns:        {"answer": "AI 的回答", "sources": [{"score": 0.85, "file": "xxx.md"}, ...]}    """    query_engine = create_query_engine(top_k=top_k)    response = query_engine.query(question)    sources = []    for node in response.source_nodes:        sources.append({            "score": round(node.score, 4) if node.score else 0,            "text": node.text[:500] if node.text else "",            "file": node.metadata.get("file_name", "unknown"),        })    if verbose:        print(f"\n检索到 {len(sources)} 个相关片段：")        for i, s in enumerate(sources):            print(f"  [{i+1}] 相关度={s['score']:.4f}  文件={s['file']}")            print(f"      {s['text'][:120]}...")    return {"answer": str(response), "sources": sources}

这条链路是怎么跑的：

你问："红绿灯模型是什么？"     ↓1. embedding("红绿灯模型是什么？") → 一个 1024 维的向量     ↓2. 在 pgvector 里找和这个向量最接近的 5 个数据块     ↓3. MetadataReplacementPostProcessor：把每个小块换成大窗口     ↓4. 拼接成 prompt：   "根据以下资料回答问题：{窗口1} {窗口2} ... 问题：红绿灯模型是什么？"     ↓5. 发给 qwen-plus → 生成回答 → 返回给你（附来源）

验证：

cd D:\qcgs && python -c "from rag.engine import query; print('engine 加载成功')"

---

8. 第六步：CLI + API 双入口

---

8.1 CLI 命令行接口

在 D:\qcgs\rag\cli.py 写入：

"""命令行入口。"""import sysimport clickfrom .engine import build_index, query# 修正在 Windows 终端输出 emoji 时的编码报错if sys.platform == "win32":    sys.stdout.reconfigure(encoding="utf-8", errors="replace")@click.group()def cli():    """QCGS RAG — 用自然语言对话你的文档。"""    pass@cli.command()@click.option("--docs-dir", default=None, help="文档目录路径")@click.option("--verbose/--quiet", default=True, help="显示进度")def build(docs_dir, verbose):    """建立（或重建）索引。"""    print("正在建立索引...")    build_index(docs_dir=docs_dir, rebuild=True, verbose=verbose)    print("完成。现在可以运行: python -m rag.run query")@cli.command()@click.argument("question", required=False)@click.option("--top-k", default=5, help="检索几条相关内容")@click.option("--verbose/--quiet", default=False, help="显示检索详情")def search(question, top_k, verbose):    """提问。"""    if question is None:        question = click.prompt("请输入问题")    result = query(question, top_k=top_k, verbose=verbose)    click.echo()    click.echo("=" * 60)    click.echo(result["answer"])    click.echo("=" * 60)    if result["sources"]:        click.echo(f"\n参考来源 ({len(result['sources'])} 条):")        for i, s in enumerate(result["sources"]):            click.echo(f"  [{i+1}] {s['file']}  (相关度: {s['score']:.4f})")if __name__ == "__main__":    cli()

8.2 API 服务

在 D:\qcgs\rag\api.py 写入：

"""REST API 服务。"""from contextlib import asynccontextmanagerfrom fastapi import FastAPI, HTTPExceptionfrom fastapi.middleware.cors import CORSMiddlewarefrom pydantic import BaseModel, Fieldfrom .config import API_HOST, API_PORT, TOP_Kfrom .engine import query as rag_query, build_index# --- 请求 / 响应格式定义 ---class QueryRequest(BaseModel):    question: str = Field(..., min_length=1, max_length=2000)    top_k: int = Field(default=5, ge=1, le=20)class QueryResponse(BaseModel):    answer: str    sources: list[dict]class HealthResponse(BaseModel):    status: str    version: strclass RebuildResponse(BaseModel):    status: str    message: str# --- 应用启动 ---@asynccontextmanagerasync def lifespan(app: FastAPI):    from .store import setup_store    setup_store()    yieldapp = FastAPI(title="QCGS RAG API", version="0.1.0", lifespan=lifespan)app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])# --- 接口 ---@app.get("/health", response_model=HealthResponse)async def health():    return HealthResponse(status="ok", version="0.1.0")@app.post("/query", response_model=QueryResponse)async def query_endpoint(req: QueryRequest):    try:        result = rag_query(req.question, top_k=req.top_k)        return QueryResponse(**result)    except Exception as e:        raise HTTPException(status_code=500, detail=str(e))@app.post("/rebuild", response_model=RebuildResponse)async def rebuild():    try:        build_index(rebuild=True, verbose=True)        return RebuildResponse(status="ok", message="索引重建成功")    except Exception as e:        raise HTTPException(status_code=500, detail=str(e))def main():    import uvicorn    uvicorn.run("rag.api:app", host=API_HOST, port=API_PORT, reload=True)if __name__ == "__main__":    main()

8.3 统一入口

在 D:\qcgs\rag\run.py 写入：

"""统一入口：python -m rag.run <命令>"""import argparseimport sysdef main():    parser = argparse.ArgumentParser(prog="qcgs-rag")    sub = parser.add_subparsers(dest="command", required=True)    sub.add_parser("build", help="建立/重建索引")    sub.add_parser("query", help="交互式提问")    q_parser = sub.add_parser("q", help="直接提问")    q_parser.add_argument("question", nargs="*", help="问题文本")    api_parser = sub.add_parser("api", help="启动 API 服务")    api_parser.add_argument("--port", type=int, default=8000)    api_parser.add_argument("--host", default="0.0.0.0")    args = parser.parse_args()    if args.command == "build":        from .cli import cli        sys.argv = ["cli", "build"]; cli()    elif args.command == "query":        from .cli import cli        sys.argv = ["cli", "search"]; cli()    elif args.command == "q":        from .cli import cli        question = " ".join(args.question)        sys.argv = ["cli", "search", question]; cli()    elif args.command == "api":        import uvicorn        from .api import app, API_HOST, API_PORT        host = args.host or API_HOST        port = args.port or API_PORT        print(f"API 服务启动: http://{host}:{port}")        print(f"  健康检查: GET  http://{host}:{port}/health")        print(f"  提问:     POST http://{host}:{port}/query")        uvicorn.run(app, host=host, port=port)if __name__ == "__main__":    main()

验证：

cd D:\qcgs && python -m rag.run --help# 应该看到 build / query / q / api 四个子命令

---

9. 第七步：第一次运行！

---

9.1 建立索引

cd D:\qcgspython -m rag.run build

你会看到类似这样的输出：

清空旧索引...正在读取文档并分块...加载了 10 个文档...  - C.md  - girl1_log.md  ...生成了 47 个数据块。正在将 47 个向量写入数据库...索引建立完成！

这一步会调用阿里云 API，CC.md 如果很大的话可能需要 1-2 分钟。 耐心等，别关终端。

9.2 提第一个问题

python -m rag.run q "红绿灯模型是什么？"

输出：

恭喜！你的 RAG 跑起来了。

9.3 启动 API（可选）

python -m rag.run api

另开一个终端测试：

curl -X POST http://localhost:8000/query \  -H "Content-Type: application/json" \  -d '{"question": "红绿灯模型是什么？"}'

---

10. 第八步：加个监控面板

---

LangSmith 是什么

想象你开了家餐厅。没有监控，你只知道"客人吃了饭走了"。有监控，你能看到：哪道菜做得最慢？哪个服务员被点单最多？哪个环节卡住了？

LangSmith 就是 RAG 的监控摄像头。每次查询它会记录：

• • 用户问了什么
• • 检索到了哪些内容（以及相关度分数）
• • LLM 花了多少秒、消耗了多少 token
• • 完整 prompt 是什么

这对调试非常有用——比如"明明文档里有答案，为什么 AI 答错了？"你一看 LangSmith，发现是检索没命中，那就该调分块策略，而不是改 prompt。

开启方式

三步，两分钟：

1. 1. 去 smith.langchain.com 注册，拿到 API Key
2. 2. 修改 .env 里的三行：

LANGSMITH_API_KEY=lsv2_pt_你的keyLANGSMITH_PROJECT=qcgs-ragLANGSMITH_TRACING=true

1. 3. 重启查询：

python -m rag.run q "测试问题"

终端会打印：

[LangSmith] 追踪已开启 — 项目: qcgs-rag

然后打开 smith.langchain.com，选 qcgs-rag 项目，就能看到刚才这次查询的完整调用链。

关掉追踪

把 LANGSMITH_TRACING=false 就行，代码不用改（get_tracer() 会自动返回一个什么都不做的假追踪器，零开销）。

---

11. 第九步：LangSmith Studio 本地 Chat UI

---

这是什么

LangSmith Studio 是一个跑在你电脑上的 Chat 调试面板。之前第八步的 LangSmith 是"云端监控"——你需要打开 smith.langchain.com 网页才能看到数据。而 Studio 更进一步：你在浏览器里直接打字提问，右侧面板实时显示检索到了哪些文档、LLM 调用花了多少秒、完整 prompt 是什么。

和第八步 LangSmith 的关系：

• • LangSmith = 云端记录每次查询 → 适合长期统计、团队共享
• • Studio = 本地实时调试 → 适合开发时改参数、看效果
• • 两者可以同时开，互不干扰

新增依赖

修改 pyproject.toml（上面第三步已经更新过了），新增了：

• • langgraph — 把 RAG 引擎包成 LangGraph 图，让 Studio 能识别
• • langgraph-cli[inmem] — 提供 langgraph dev 命令
• • langchain-core — LangGraph 底层消息格式

然后安装：

cd D:\qcgs\ragpip install -e .

新建 graph.py — 把 RAG 包装成 LangGraph 图

在 D:\qcgs\rag\graph.py 写入：

"""LangGraph 包装 — LangSmith Studio 通过这个文件发现和调用你的 RAG。启动 Studio 后，每次在 Chat UI 里打字，都会走这里的 _rag_node 函数。"""from langgraph.graph import StateGraph, MessagesState, START, ENDfrom langchain_core.messages import AIMessage# Studio 直接加载这个文件时没有 package 上下文，# 所以两套 import 都要支持try:    from .engine import query as rag_query    from .config import TOP_Kexcept ImportError:    from rag.engine import query as rag_query    from rag.config import TOP_Kdef _rag_node(state: MessagesState) -> dict:    """检索 + 生成回答，附上参考来源。"""    messages = state["messages"]    question = messages[-1].content    result = rag_query(question, top_k=TOP_K)    answer = result["answer"]    if result["sources"]:        lines = ["\n\n**参考来源：**"]        for i, s in enumerate(result["sources"]):            lines.append(f"  [{i + 1}] {s['file']} (相关度: {s['score']:.2f})")        answer += "\n".join(lines)    return {"messages": [AIMessage(content=answer)]}def create_graph():         builder = StateGraph(MessagesState)         builder.add_node("rag", _rag_node)         builder.add_edge(START, "rag")         builder.add_edge("rag", END)         # ⚠️ Studio 自带持久化，不要手动传 checkpointer         return builder.compile()graph = create_graph()

新建 langgraph.json — 告诉 Studio 去哪找图

在 项目根目录 D:\qcgs\langgraph.json 写入：

{  "dependencies":["./rag"],"graphs":{    "rag":"./rag/graph.py:graph"},"env":"rag/.env","python_version":"3.13"}

字段	作用
dependencies	告诉 Studio 去哪找 Python 包（./rag = rag 目录下的 pyproject.toml）
graphs	rag 是图的名字（Studio 里显示的 assistant 名称），后面是文件路径 :变量名
env	环境变量文件位置，Studio 启动时会自动加载

启动

cd D:\qcgsPYTHONUTF8=1 langgraph dev --port 2024

PYTHONUTF8=1 是 Windows 必须加的，否则遇到 emoji 就崩溃。

看到这个就说明成功了：

Welcome to╦  ┌─┐┌┐┌┌─┐╔═╗┬─┐┌─┐┌─┐┬ ┬║  ├─┤││││ ┬║ ╦├┬┘├─┤├─┘├─┤╩═╝┴ ┴┘└┘└─┘╚═╝┴└─┴ ┴┴  ┴ ┴- 🚀 API: http://127.0.0.1:2024- 🎨 Studio UI: https://smith.langchain.com/studio/?baseUrl=http://127.0.0.1:2024

浏览器打开上面那个 Studio UI 链接，左侧选 rag，就可以在 Chat UI 里对话了。右侧面板会实时展示每次查询的 trace。

数据安全：studio 网页本身托管在 LangChain 的 CDN，但你的对话内容、文档数据、API Key 全都只在本地 127.0.0.1:2024 传输，不会上传到 LangChain 服务器。

关掉

终端按 Ctrl+C 即可。

---

12. 第十步：Phoenix 零上报本地监控

---

这是什么

LangSmith Studio 的 Chat UI 很方便，但它的网页是从 smith.langchain.com 加载的。如果你想要**完全不依赖任何外部服务器（连网页 CDN 都自己跑）**的本地监控面板，用 Arize Phoenix。

Phoenix 是一个开源的 LLM 可观测性工具：

• • 所有数据存在你电脑的 SQLite 文件里（C:\Users\你的用户名\.phoenix）
• • 自带 Web UI，打开 http://localhost:6006 就能看
• • 能追踪每次查询的完整 Span 树：embedding 耗时 → 检索耗时 → LLM 调用耗时
• • 零外部网络依赖（除了你调用阿里云 API 的部分）

怎么加进去的（已有代码，不需再写）

上面第 3～4 步的代码里已经包含了 Phoenix 配置，回顾一下关键部分：

config.py 新增：

PHOENIX_ENABLED = os.getenv("PHOENIX_ENABLED", "false").lower() == "true"PHOENIX_ENDPOINT = os.getenv("PHOENIX_ENDPOINT", "http://localhost:6006/v1/traces")def setup_phoenix():    if not PHOENIX_ENABLED:        return    from phoenix.otel import register    from openinference.instrumentation.llama_index import LlamaIndexInstrumentor    register(project_name="qcgs-rag", endpoint=PHOENIX_ENDPOINT, batch=True)    LlamaIndexInstrumentor().instrument()

init.py 在 .env 加载后立即调用：

from .config import setup_phoenix as _setup_phoenix_setup_phoenix()  # 必须在 LlamaIndex 被 import 之前执行

启动

需要两个终端：

终端 1：启动 Phoenix 服务

cd D:\qcgsPYTHONUTF8=1 python -m phoenix.server.main serve --port 6006

看到 Started server process 就说明起来了，打开 http://localhost:6006 确认能看到 Phoenix 的界面。

终端 2：启动 LangGraph Studio（带 Phoenix 追踪）

先改 .env：

PHOENIX_ENABLED=true

然后：

cd D:\qcgsPYTHONUTF8=1 langgraph dev --port 2024

日志里会打印：

[Phoenix] 本地追踪已开启 — UI: http://localhost:6006

效果

1. 1. 在 Studio Chat UI（https://smith.langchain.com/studio/?baseUrl=http://127.0.0.1:2024）里打字提问
2. 2. 切到 Phoenix 页面（http://localhost:6006）
3. 3. 选 qcgs-rag 项目
4. 4. 每次查询会以 Span 树的形式展示：根 span → embedding 子 span → retrieval 子 span → LLM 子 span，每个带耗时

关掉

两个终端各自 Ctrl+C。

三种监控模式总结

模式	.env 配置	数据去向	适用场景
无监控	LANGSMITH_TRACING=false PHOENIX_ENABLED=false	无	生产环境、不想有任何追踪开销
纯 LangSmith	LANGSMITH_TRACING=true	云端 smith.langchain.com	团队协作、长期统计
纯 Phoenix	PHOENIX_ENABLED=true	本地 SQLite	个人开发调试、数据不出电脑
双开	两个都 true	本地 + 云端各一份	开发期全面监控

---

13. 常见报错速查表

---

这 6 个坑是我们在真实搭建中一个个踩过来的，每个都折腾了至少 10 分钟。

#	报错信息	为什么	怎么修
1	batch size is invalid, it should not be larger than 10	阿里云 embedding API 每次最多接收 10 条文本	在 DashScopeEmbedding() 里加 embed_batch_size=10
2	Range of input length should be [1, 8192]	markdown 代码块被当作一整句话，超过 8192 token	用 SentenceSplitter(chunk_size=2048) 包装分句器
3	KeyError: 'hnsw_ef_search'	pgvector 的 HNSW 参数缺了带 hnsw_ 前缀的版本	hnsw_kwargs 里同时写 hnsw_ef_search、hnsw_m、hnsw_ef_construction
4	连接数慢慢涨，最后数据库连不上	用 conn = psycopg2.connect() 不用 with，异常时连接泄露	全部改成 with psycopg2.connect(...) as conn:
5	ModuleNotFoundError: No module named 'langsmith'	用了系统 Python 而不是 venv 里的	pip install langsmith 到对应的 Python
6	UnicodeEncodeError: 'gbk' codec can't encode	Windows 终端默认 GBK 编码，AI 回答里有 emoji	在 cli.py 顶部加 sys.stdout.reconfigure(encoding="utf-8")
7	langgraph dev 报 No dependencies found in config	langgraph.json 缺少 dependencies 字段	加上 "dependencies": ["./rag"]
8	langgraph dev 报 attempted relative import with no known parent package	相对导入在 Studio 加载时没有 package 上下文	用 try/except 同时支持相对和绝对导入（见 graph.py 示例）
9	langgraph dev 报 includes a custom checkpointer	手动传了 MemorySaver，但 Studio 自带持久化	删掉 checkpointer=memory，直接 builder.compile()
10	Phoenix 页面打开但没有 trace 数据	PHOENIX_ENABLED=false 或者 Phoenix 服务没启动	检查 .env 里 PHOENIX_ENABLED=true，确认 6006 端口已启动

---

14. 换成你自己的

---

换成你自己的文档

修改 config.py 里的 DOCS_DIR，或者直接设环境变量：

set DOCS_DIR=C:\my-docspython -m rag.run build

换成 OpenAI

# config.py 改模型名LLM_MODEL = "gpt-4o"EMBEDDING_MODEL = "text-embedding-3-large"# loader.py / engine.py 替换 importfrom llama_index.embeddings.openai import OpenAIEmbeddingfrom llama_index.llms.openai import OpenAI

换成不需要数据库的本地向量库

如果不想装 PostgreSQL，可以换成 Chroma（一个文件型向量库）：

# store.py 改为from llama_index.vector_stores.chroma import ChromaVectorStoreimport chromadbdef get_vector_store():    chroma_client = chromadb.PersistentClient(path="./chroma_db")    chroma_collection = chroma_client.get_or_create_collection("qcgs")    return ChromaVectorStore(chroma_collection=chroma_collection)

调检索质量

如果发现 AI 的回答不太对，先调这几个参数（都在 config.py）：

参数	作用	太小	太大
SENTENCE_WINDOW_SIZE	给 AI 看的上下文窗口	断章取义	噪音多、token 浪费
TOP_K	检索几条相关内容	信息不足	可能混入不相关的内容
SENTENCE_CHUNK_SIZE	每块的 token 数	语义破碎	检索精度下降

参考调整方向：

• • AI 答非所问 → 加大 TOP_K、检查分块是否合理
• • AI 回答太泛 → 加大 SENTENCE_WINDOW_SIZE、减小 SENTENCE_CHUNK_SIZE
• • 检索太慢 → 减少 TOP_K
• • 总是找不到答案 → 检查文档是否被 exclude 过滤掉了

---

2026年AI行业最大的机会，毫无疑问就在应用层！

字节跳动已有7个团队全速布局Agent

大模型岗位暴增69%，年薪破百万！

腾讯、京东、百度开放招聘技术岗，80%与AI相关……

如今，超过60%的企业都在推进AI产品落地，而真正能交付项目的 大模型应用开发工程师 **，**却极度稀缺！

落地AI应用绝对不是写几个prompt，调几个API就能搞定的，企业真正需要的，是能搞定这三项核心能力的人：

✅RAG：融入外部信息，修正模型输出，给模型装靠谱大脑

✅Agent智能体：让AI自主干活，通过工具调用（Tools）环境交互，多步推理完成复杂任务。比如做智能客服等等……

✅微调：针对特定任务优化，让模型适配业务

目前，脉脉上有超过1000家企业发布大模型相关岗位，人工智能岗平均月薪7.8w！实习生日薪高达4000！远超其他行业收入水平！

技术的稀缺性，才是你「值钱」的关键！

具备AI能力的程序员，比传统开发高出不止一截！有的人早就转行AI方向，拿到百万年薪！👇🏻👇🏻

AI浪潮，正在重构程序员的核心竞争力！现在入场，仍是最佳时机！

我把大模型的学习全流程已经整理📚好了！抓住AI时代风口，轻松解锁职业新可能，希望大家都能把握机遇，实现薪资/职业跃迁～

这份完整版的大模型 AI 学习资料已经上传CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】

⭐️从大模型微调到AI Agent智能体搭建

剖析AI技术的应用场景，用实战经验落地AI技术。从GPT到最火的开源模型，让你从容面对AI技术革新！

大模型微调

• 掌握主流大模型（如DeepSeek、Qwen等）的微调技术，针对特定场景优化模型性能。

• 学习如何利用领域数据（如制造、医药、金融等）进行模型定制，提升任务准确性和效率。

RAG应用开发

• 深入理解检索增强生成（Retrieval-Augmented Generation, RAG）技术，构建高效的知识检索与生成系统。
• 应用于垂类场景（如法律文档分析、医疗诊断辅助、金融报告生成等），实现精准信息提取与内容生成。

AI Agent智能体搭建

• 学习如何设计和开发AI Agent，实现多任务协同、自主决策和复杂问题解决。
• 构建垂类场景下的智能助手（如制造业中的设备故障诊断Agent、金融领域的投资分析Agent等）。

如果你也有以下诉求：

快速链接产品/业务团队，参与前沿项目

构建技术壁垒，从竞争者中脱颖而出

避开35岁裁员危险期，顺利拿下高薪岗

迭代技术水平，延长未来20年的新职业发展！

……

那这节课你一定要来听！

因为，留给普通程序员的时间真的不多了！

立即扫码，即可免费预约

「AI技术原理 + 实战应用 + 职业发展」

「大模型应用开发实战公开课」

👇👇

👍🏻还有靠谱的内推机会+直聘权益！！

完课后赠送：大模型应用案例集、AI商业落地白皮书