> ## Documentation Index
> Fetch the complete documentation index at: https://docs.textin.com/llms.txt
> Use this file to discover all available pages before exploring further.

# xParse + LangGraph 构建 Agentic RAG

> 使用 xParse SDK + LangGraph 构建 Agentic RAG，智能重写问题、检索与回答，实现更准确的企业知识问答

<Tip>
  本教程将带您了解如何使用 [xParse SDK](/xparse/v1/sdk-python) + LangGraph 构建 Agentic RAG，智能重写问题、检索与回答，实现更准确的企业知识问答。
</Tip>

## 什么是 Agentic RAG？

Agentic RAG（Retrieval-Augmented Generation）是一种结合信息检索和生成式 AI 的技术。与传统的 RAG 不同，Agentic RAG 通过智能决策机制，能够：

1. **智能判断**：判断问题是否需要检索，还是可以直接回答
2. **问题重写**：当检索结果不相关时，自动重写问题以获取更好的结果
3. **相关性评估**：评估检索到的文档片段是否与问题相关
4. **迭代优化**：通过多轮检索和重写，逐步优化答案质量

Agentic RAG 的核心流程包括：

1. **文档处理**：将非结构化文档转换为向量表示
2. **向量存储**：将向量数据存储到向量数据库
3. **智能检索**：根据用户问题智能检索相关文档片段
4. **问题重写**：当检索结果不相关时，重写问题再次检索
5. **生成回答**：基于检索到的相关上下文生成高质量答案

## Agentic RAG 工作流程

```
用户问题："如何配置数据库连接池？"
    ↓
[LangGraph 工作流]  ←  [xParse SDK]
    ↓
[判断节点] 是否需要检索？
    ├─ 是 → [检索节点] 向量检索
    │         ↓ 
    │   [评估节点] 检索结果是否相关？
    │         ├─ 相关 → [生成节点] 生成答案
    │         └─ 不相关 → [重写节点] 重写问题
    │                           ↓
    │                    [检索节点] 再次检索
    │                           ↓
    │                    [评估节点] 再次评估
    │                           ↓
    │                    [生成节点] 生成答案
    └─ 否 → [生成节点] 直接生成答案
```

## 环境准备

首先安装必要的依赖：

```bash theme={null}
python -m venv .venv && source .venv/bin/activate
pip install xparse-client langchain langchain-community langchain-core \
            langchain-text-splitters langgraph langchain-milvus python-dotenv
```

创建 `.env` 文件存储配置：

```bash theme={null}
# .env
TEXTIN_APP_ID=your-app-id
TEXTIN_SECRET_CODE=your-secret-code
MILVUS_DB_PATH=./agentic_rag_vectors.db
DASHSCOPE_API_KEY=your-dashscope-key
```

> 提示：`TEXTIN_APP_ID` 与 `TEXTIN_SECRET_CODE` 参考 [API Key](/xparse/api-key)，请登录 [Textin 工作台](https://www.textin.com/console/dashboard/setting) 获取。示例中使用 `通义千问` 的大模型能力，其他模型用法类似。

下面我们将分步骤构建 Agentic RAG 系统。首先导入必要的库：

```python theme={null}
import os
import glob
from typing import TypedDict, Annotated
from dotenv import load_dotenv
from xparse_client import XParseClient
from langchain_text_splitters import MarkdownHeaderTextSplitter, RecursiveCharacterTextSplitter
from langchain_milvus import Milvus
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_community.chat_models import ChatTongyi
from langchain_core.documents import Document
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langchain_core.messages import HumanMessage, BaseMessage

load_dotenv()
```

### Step 1: 使用 xParse SDK 构建知识库

首先，我们需要使用 [xParse SDK](/xparse/v1/sdk-python) 解析文档，再通过 LangChain 进行分块和向量化，最终存入 Milvus 向量数据库。这是知识库构建的基础步骤。

初始化 xParse 客户端并构建知识库：

```python theme={null}
client = XParseClient()

def build_knowledge_base():
    """构建知识库"""
    print("开始构建知识库...")
    
    docs_dir = "./knowledge_base"
    headers_to_split_on = [("#", "header1"), ("##", "header2"), ("###", "header3")]
    markdown_splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers_to_split_on)
    text_splitter = RecursiveCharacterTextSplitter(chunk_size=1500, chunk_overlap=80)
    
    all_chunks = []
    for file_path in glob.glob(os.path.join(docs_dir, "*")):
        if not os.path.isfile(file_path):
            continue
        with open(file_path, "rb") as f:
            result = client.parse.run(file=f, filename=os.path.basename(file_path))
        md_docs = markdown_splitter.split_text(result.markdown)
        for doc in md_docs:
            doc.metadata["filename"] = os.path.basename(file_path)
        chunks = text_splitter.split_documents(md_docs)
        all_chunks.extend(chunks)
    
    embedding = DashScopeEmbeddings(model="text-embedding-v4")
    Milvus.from_documents(
        documents=all_chunks,
        embedding=embedding,
        collection_name="agentic_rag_docs",
        connection_args={"uri": os.getenv("MILVUS_DB_PATH")},
    )
    print("知识库构建完成！")
```

### Step 2: 初始化向量数据库和大模型

接下来，我们需要初始化向量数据库和大模型。**重要**：向量数据库必须使用与构建知识库时相同的 embedding 模型，以确保语义空间一致。

```python theme={null}
embedding = DashScopeEmbeddings(model="text-embedding-v4")
vector_store = Milvus(
    embedding_function=embedding,
    collection_name="agentic_rag_docs",
    connection_args={"uri": os.getenv("MILVUS_DB_PATH")},
)

llm = ChatTongyi(
    model="qwen-max",
    top_p=0.8,
    dashscope_api_key=os.getenv("DASHSCOPE_API_KEY")
)
```

### Step 3: 定义状态结构

LangGraph 使用状态来管理工作流中的数据流。我们需要定义一个 `GraphState` 来存储工作流中的各种信息：

```python theme={null}
class GraphState(TypedDict):
    """工作流状态"""
    messages: Annotated[list[BaseMessage], add_messages]
    question: str  # 原始问题
    rewritten_question: str  # 重写后的问题
    documents: list  # 检索到的文档
    generation: str  # 生成的答案
    next: str  # 下一步操作
    retrieval_count: int  # 检索次数
```

状态字段说明：

* `question`：用户提出的原始问题
* `rewritten_question`：重写后的问题（用于优化检索）
* `documents`：检索到的相关文档片段
* `generation`：最终生成的答案
* `next`：指示下一步应该执行哪个节点
* `retrieval_count`：检索次数（用于防止无限循环）

### Step 4: 定义节点函数

工作流由多个节点组成，每个节点负责特定的任务。让我们逐个实现这些节点：

#### 4.1 判断是否需要检索

`should_retrieve` 节点使用 LLM 判断问题是否需要从知识库检索信息：

```python theme={null}
def should_retrieve(state: GraphState) -> GraphState:
    """判断是否需要检索"""
    question = state["question"]
    
    prompt = f"""判断以下问题是否需要从知识库中检索信息才能回答。

            问题：{question}

            如果问题需要特定的文档、数据或知识库信息才能回答，返回 "retrieve"。
            如果问题是一般性对话、问候或不需要特定信息的简单问题，返回 "generate"。

            只返回 "retrieve" 或 "generate"，不要返回其他内容。"""
    
    response = llm.invoke([HumanMessage(content=prompt)])
    decision = response.content.strip().lower()
    
    next_step = "retrieve" if "retrieve" in decision else "generate"
    
    return {
        **state,
        "next": next_step
    }
```

#### 4.2 检索相关文档

`retrieve` 节点使用向量检索在知识库中查找相关内容：

```python theme={null}
def retrieve(state: GraphState) -> GraphState:
    """检索相关文档"""
    question = state.get("rewritten_question") or state["question"]
    retrieval_count = state.get("retrieval_count", 0)
    
    docs = vector_store.similarity_search(question, k=5)
    
    documents = []
    for doc in docs:
        documents.append({
            "content": doc.page_content,
            "metadata": doc.metadata
        })
    
    return {
        **state,
        "documents": documents,
        "retrieval_count": retrieval_count + 1
    }
```

注意：这里优先使用 `rewritten_question`（如果存在），否则使用原始问题。每次检索后，`retrieval_count` 会增加 1。

#### 4.3 评估检索结果的相关性

`grade_documents` 节点评估检索到的文档是否与问题相关：

```python theme={null}
def grade_documents(state: GraphState) -> GraphState:
    """评估检索结果的相关性"""
    question = state.get("rewritten_question") or state["question"]
    documents = state["documents"]
    retrieval_count = state.get("retrieval_count", 0)
    
    if retrieval_count >= 2:
        return {
            **state,
            "next": "generate"
        }
    
    if not documents:
        return {
            **state,
            "next": "rewrite"
        }
    
    docs_text = "\n\n".join([
        f"文档 {i+1}:\n{doc['content'][:300]}..."
        for i, doc in enumerate(documents[:3])
    ])
    
    prompt = f"""评估以下检索到的文档是否与问题相关。

            问题：{question}

            检索到的文档：
            {docs_text}

            如果文档与问题高度相关，能够回答问题，返回 "generate"。
            如果文档与问题不相关或相关性很低，返回 "rewrite"。

            只返回 "generate" 或 "rewrite"，不要返回其他内容。"""
    
    response = llm.invoke([HumanMessage(content=prompt)])
    decision = response.content.strip().lower()
    
    next_step = "generate" if "generate" in decision else "rewrite"
    
    return {
        **state,
        "next": next_step
    }
```

如果文档不相关，将触发问题重写。为了避免无限循环，当检索次数达到 2 次时，即使文档不相关也会强制生成答案。

#### 4.4 重写问题

`rewrite_question` 节点基于检索结果或原始问题，生成更优化的查询：

```python theme={null}
def rewrite_question(state: GraphState) -> GraphState:
    """重写问题"""
    question = state["question"]
    documents = state.get("documents", [])
    
    if documents:
        docs_summary = "\n".join([
            f"- {doc['content'][:200]}..."
            for doc in documents[:2]
        ])
        prompt = f"""原始问题：{question}

                当前检索到的文档摘要：
                {docs_summary}

                这些文档与问题不够相关。请重写问题，使其能够更好地匹配知识库中的内容。
                重写时应该：
                1. 保持问题的核心意图
                2. 使用更具体的关键词
                3. 考虑知识库可能使用的术语

                只返回重写后的问题，不要返回其他内容。"""
    else:
        prompt = f"""原始问题：{question}

                请重写这个问题，使其更具体、更清晰，便于在知识库中检索相关信息。
                重写时应该：
                1. 保持问题的核心意图
                2. 使用更具体的关键词
                3. 考虑知识库可能使用的术语

                只返回重写后的问题，不要返回其他内容。"""
    
    response = llm.invoke([HumanMessage(content=prompt)])
    rewritten = response.content.strip()
    
    return {
        **state,
        "rewritten_question": rewritten
    }
```

#### 4.5 生成答案

`generate` 节点基于检索结果或直接生成答案：

```python theme={null}
def generate(state: GraphState) -> GraphState:
    """生成答案"""
    question = state["question"]
    documents = state.get("documents", [])
    
    if documents:
        context = "\n\n".join([
            f"文档来源：{doc['metadata'].get('filename', '未知')}\n内容：{doc['content']}"
            for i, doc in enumerate(documents)
        ])
        
        prompt = f"""基于以下文档内容回答用户问题。

                文档内容：
                {context}

                用户问题：{question}

                请基于文档内容回答问题。如果文档中没有相关信息，请说明。
                在回答中引用具体的文档来源。"""
    else:
        prompt = f"""回答以下问题：{question}"""
    
    response = llm.invoke([HumanMessage(content=prompt)])
    
    return {
        **state,
        "generation": response.content
    }
```

### Step 5: 构建 LangGraph 工作流

现在我们将所有节点组合成一个完整的工作流：

```python theme={null}
workflow = StateGraph(GraphState)

workflow.add_node("should_retrieve", should_retrieve)
workflow.add_node("retrieve", retrieve)
workflow.add_node("grade_documents", grade_documents)
workflow.add_node("rewrite_question", rewrite_question)
workflow.add_node("generate", generate)

workflow.set_entry_point("should_retrieve")

workflow.add_conditional_edges(
    "should_retrieve",
    lambda state: state.get("next", "generate"),
    {
        "retrieve": "retrieve",
        "generate": "generate"
    }
)

workflow.add_edge("retrieve", "grade_documents")

workflow.add_conditional_edges(
    "grade_documents",
    lambda state: state.get("next", "generate"),
    {
        "generate": "generate",
        "rewrite": "rewrite_question"
    }
)

workflow.add_edge("rewrite_question", "retrieve")
workflow.add_edge("generate", END)

app = workflow.compile()
```

工作流逻辑：

1. 从 `should_retrieve` 开始
2. 如果需要检索，进入 `retrieve` → `grade_documents`
3. 如果文档相关，进入 `generate`；如果不相关，进入 `rewrite_question` → `retrieve`（循环）
4. 如果不需要检索，直接进入 `generate`

### Step 6: 使用示例

创建一个便捷的提问函数：

```python theme={null}
def ask_question(question: str) -> str:
    """提问并获取答案"""
    initial_state = {
        "messages": [HumanMessage(content=question)],
        "question": question,
        "rewritten_question": "",
        "documents": [],
        "generation": "",
        "next": "",
        "retrieval_count": 0
    }
    
    result = app.invoke(initial_state)
    
    return result["generation"]
```

使用示例：

```python theme={null}
if __name__ == "__main__":
    build_knowledge_base()
    
    questions = [
        "如何配置数据库连接池？",
        "产品的定价策略是什么？",
        "你好"
    ]
    
    for question in questions:
        print("=" * 60)
        print(f"问题：{question}")
        print("=" * 60)
        answer = ask_question(question)
        print(f"回答：{answer}\n")
```

## 完整代码示例

下面是一个完整的、可以直接运行的示例：

```python expandable theme={null}
import os
import glob
from typing import TypedDict, Annotated, Literal
from dotenv import load_dotenv
from xparse_client import XParseClient
from langchain_text_splitters import MarkdownHeaderTextSplitter, RecursiveCharacterTextSplitter
from langchain_milvus import Milvus
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_community.chat_models import ChatTongyi
from langchain_core.documents import Document
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langchain_core.messages import HumanMessage, AIMessage, BaseMessage

load_dotenv()

# ========== Step 1: 使用 xParse SDK 构建知识库 ==========

client = XParseClient()

def build_knowledge_base():
    """构建知识库"""
    print("开始构建知识库...")
    
    docs_dir = "./knowledge_base"
    headers_to_split_on = [("#", "header1"), ("##", "header2"), ("###", "header3")]
    markdown_splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers_to_split_on)
    text_splitter = RecursiveCharacterTextSplitter(chunk_size=1500, chunk_overlap=80)
    
    all_chunks = []
    for file_path in glob.glob(os.path.join(docs_dir, "*")):
        if not os.path.isfile(file_path):
            continue
        with open(file_path, "rb") as f:
            result = client.parse.run(file=f, filename=os.path.basename(file_path))
        md_docs = markdown_splitter.split_text(result.markdown)
        for doc in md_docs:
            doc.metadata["filename"] = os.path.basename(file_path)
        chunks = text_splitter.split_documents(md_docs)
        all_chunks.extend(chunks)
    
    embedding = DashScopeEmbeddings(model="text-embedding-v4")
    Milvus.from_documents(
        documents=all_chunks,
        embedding=embedding,
        collection_name="agentic_rag_docs",
        connection_args={"uri": os.getenv("MILVUS_DB_PATH")},
    )
    print("知识库构建完成！")

# ========== Step 2: 初始化向量数据库和大模型 ==========

embedding = DashScopeEmbeddings(model="text-embedding-v4")
vector_store = Milvus(
    embedding_function=embedding,
    collection_name="agentic_rag_docs",
    connection_args={"uri": os.getenv("MILVUS_DB_PATH")},
)

llm = ChatTongyi(
    model="qwen-max",
    top_p=0.8,
    dashscope_api_key=os.getenv("DASHSCOPE_API_KEY")
)

# ========== Step 3: 定义状态结构 ==========

class GraphState(TypedDict):
    """工作流状态"""
    messages: Annotated[list[BaseMessage], add_messages]
    question: str  # 原始问题
    rewritten_question: str  # 重写后的问题
    documents: list  # 检索到的文档
    generation: str  # 生成的答案
    next: str  # 下一步操作
    retrieval_count: int  # 检索次数

# ========== Step 4: 定义节点函数 ==========

def should_retrieve(state: GraphState) -> GraphState:
    """判断是否需要检索"""
    question = state["question"]
    
    prompt = f"""判断以下问题是否需要从知识库中检索信息才能回答。

            问题：{question}

            如果问题需要特定的文档、数据或知识库信息才能回答，返回 "retrieve"。
            如果问题是一般性对话、问候或不需要特定信息的简单问题，返回 "generate"。

            只返回 "retrieve" 或 "generate"，不要返回其他内容。"""
    
    response = llm.invoke([HumanMessage(content=prompt)])
    decision = response.content.strip().lower()
    
    next_step = "retrieve" if "retrieve" in decision else "generate"
    
    return {
        **state,
        "next": next_step
    }

def retrieve(state: GraphState) -> GraphState:
    """检索相关文档"""
    question = state.get("rewritten_question") or state["question"]
    retrieval_count = state.get("retrieval_count", 0)
    
    docs = vector_store.similarity_search(question, k=5)
    
    documents = []
    for doc in docs:
        documents.append({
            "content": doc.page_content,
            "metadata": doc.metadata
        })
    
    return {
        **state,
        "documents": documents,
        "retrieval_count": retrieval_count + 1
    }

def grade_documents(state: GraphState) -> GraphState:
    """评估检索结果的相关性"""
    question = state.get("rewritten_question") or state["question"]
    documents = state["documents"]
    retrieval_count = state.get("retrieval_count", 0)
    
    if retrieval_count >= 2:
        return {
            **state,
            "next": "generate"
        }
    
    if not documents:
        return {
            **state,
            "next": "rewrite"
        }
    
    docs_text = "\n\n".join([
        f"文档 {i+1}:\n{doc['content'][:300]}..."
        for i, doc in enumerate(documents[:3])
    ])
    
    prompt = f"""评估以下检索到的文档是否与问题相关。

            问题：{question}

            检索到的文档：
            {docs_text}

            如果文档与问题高度相关，能够回答问题，返回 "generate"。
            如果文档与问题不相关或相关性很低，返回 "rewrite"。

            只返回 "generate" 或 "rewrite"，不要返回其他内容。"""
    
    response = llm.invoke([HumanMessage(content=prompt)])
    decision = response.content.strip().lower()
    
    next_step = "generate" if "generate" in decision else "rewrite"
    
    return {
        **state,
        "next": next_step
    }

def rewrite_question(state: GraphState) -> GraphState:
    """重写问题"""
    question = state["question"]
    documents = state.get("documents", [])
    
    previous_rewrite = state.get("rewritten_question", "")
    
    if documents:
        docs_summary = "\n".join([
            f"- {doc['content'][:200]}..."
            for doc in documents[:2]
        ])
        prompt = f"""原始问题：{question}

                当前检索到的文档摘要：
                {docs_summary}

                这些文档与问题不够相关。请重写问题，使其能够更好地匹配知识库中的内容。
                重写时应该：
                1. 保持问题的核心意图
                2. 使用更具体的关键词
                3. 考虑知识库可能使用的术语

                只返回重写后的问题，不要返回其他内容。"""
    else:
        prompt = f"""原始问题：{question}

                请重写这个问题，使其更具体、更清晰，便于在知识库中检索相关信息。
                重写时应该：
                1. 保持问题的核心意图
                2. 使用更具体的关键词
                3. 考虑知识库可能使用的术语

                只返回重写后的问题，不要返回其他内容。"""
    
    response = llm.invoke([HumanMessage(content=prompt)])
    rewritten = response.content.strip()
    
    return {
        **state,
        "rewritten_question": rewritten
    }

def generate(state: GraphState) -> GraphState:
    """生成答案"""
    question = state["question"]
    documents = state.get("documents", [])
    
    if documents:
        context = "\n\n".join([
            f"文档来源：{doc['metadata'].get('filename', '未知')}\n内容：{doc['content']}"
            for i, doc in enumerate(documents)
        ])
        
        prompt = f"""基于以下文档内容回答用户问题。

                文档内容：
                {context}

                用户问题：{question}

                请基于文档内容回答问题。如果文档中没有相关信息，请说明。
                在回答中引用具体的文档来源。"""
    else:
        prompt = f"""回答以下问题：{question}"""
    
    response = llm.invoke([HumanMessage(content=prompt)])
    
    return {
        **state,
        "generation": response.content
    }

# ========== Step 5: 构建 LangGraph 工作流 ==========

workflow = StateGraph(GraphState)

workflow.add_node("should_retrieve", should_retrieve)
workflow.add_node("retrieve", retrieve)
workflow.add_node("grade_documents", grade_documents)
workflow.add_node("rewrite_question", rewrite_question)
workflow.add_node("generate", generate)

workflow.set_entry_point("should_retrieve")

workflow.add_conditional_edges(
    "should_retrieve",
    lambda state: state.get("next", "generate"),
    {
        "retrieve": "retrieve",
        "generate": "generate"
    }
)

workflow.add_edge("retrieve", "grade_documents")

workflow.add_conditional_edges(
    "grade_documents",
    lambda state: state.get("next", "generate"),
    {
        "generate": "generate",
        "rewrite": "rewrite_question"
    }
)

workflow.add_edge("rewrite_question", "retrieve")
workflow.add_edge("generate", END)

app = workflow.compile()

# ========== Step 6: 使用示例 ==========

def ask_question(question: str) -> str:
    """提问并获取答案"""
    initial_state = {
        "messages": [HumanMessage(content=question)],
        "question": question,
        "rewritten_question": "",
        "documents": [],
        "generation": "",
        "next": "",
        "retrieval_count": 0
    }
    
    result = app.invoke(initial_state)
    
    return result["generation"]

if __name__ == "__main__":
    build_knowledge_base()
    
    questions = [
        "如何配置数据库连接池？",
        "产品的定价策略是什么？",
        "你好"
    ]
    
    for question in questions:
        print("=" * 60)
        print(f"问题：{question}")
        print("=" * 60)
        answer = ask_question(question)
        print(f"回答：{answer}\n")
```

## 与普通 RAG 的区别

### 普通 RAG

```
用户问题 → 向量检索 → 生成答案
```

### Agentic RAG

```
用户问题 → 判断是否需要检索
         ├─ 需要 → 检索 → 评估相关性
         │         ├─ 相关 → 生成答案
         │         └─ 不相关 → 重写问题 → 重新检索 → ...
         └─ 不需要 → 直接生成答案
```

**核心优势**：

1. **智能决策**：自动判断是否需要检索，避免不必要的检索
2. **质量保证**：评估检索结果的相关性，确保答案质量
3. **迭代优化**：通过问题重写，逐步优化检索结果
4. **灵活应对**：能够处理简单问题和复杂问题

## 实际应用场景

### 场景 1: 企业知识库问答

**需求**：员工可以通过自然语言提问，快速找到产品文档、技术文档等信息。

**实现**：

* 使用 xParse SDK 解析企业文档，通过 LangChain 分块和向量化
* Agentic RAG 自动判断问题类型
* 智能检索和重写，确保找到最相关的信息

### 场景 2: 客服助手

**需求**：客服团队需要快速回答客户问题，从 FAQ 和产品手册中找到答案。

**实现**：

* 将 FAQ 和产品手册存入知识库
* Agentic RAG 能够：
  * 识别简单问候，直接回答
  * 识别需要检索的问题，智能检索
  * 当检索结果不相关时，自动优化查询

### 场景 3: 技术文档问答

**需求**：开发者可以通过自然语言提问，快速找到 API 文档、使用示例等。

**实现**：

* 使用 xParse SDK 解析文档后，通过 MarkdownHeaderTextSplitter 按标题层级分块，保持文档结构
* Agentic RAG 能够理解技术术语，重写查询以匹配文档中的关键词

## 进阶优化

### 1. 添加检索次数限制

避免无限循环重写和检索：

```python theme={null}
class GraphState(TypedDict):
    # ... 其他字段
    retrieval_count: int  # 检索次数

def grade_documents(state: GraphState) -> GraphState:
    """评估检索结果的相关性"""
    if state.get("retrieval_count", 0) >= 2:
        return {
            **state,
            "next": "generate"
        }
    
    # ... 原有的评估逻辑
```

### 2. 添加对话历史

支持多轮对话：

```python theme={null}
class GraphState(TypedDict):
    # ... 其他字段
    chat_history: list[BaseMessage]  # 对话历史

def generate(state: GraphState) -> GraphState:
    """生成答案"""
    messages = state.get("chat_history", []) + [
        HumanMessage(content=state["question"])
    ]
    # ... 生成逻辑
```

### 3. 混合检索策略

结合向量检索和关键词检索：

```python theme={null}
def retrieve(state: GraphState) -> GraphState:
    """检索相关文档"""
    question = state.get("rewritten_question") or state["question"]
    
    vector_docs = vector_store.similarity_search(question, k=3)
    
    # keyword_docs = keyword_search(question, k=2)
    
    documents = []
    for doc in vector_docs:
        documents.append({
            "content": doc.page_content,
            "metadata": doc.metadata,
            "source": "vector"
        })
    
    return {
        **state,
        "documents": documents
    }
```

### 4. 添加引用来源

在答案中标注信息来源：

```python theme={null}
def generate(state: GraphState) -> GraphState:
    """生成答案"""
    # ... 生成逻辑
    
    if documents:
        sources = [
            doc['metadata'].get('filename', '未知')
            for doc in documents
        ]
        generation_with_sources = f"{generation}\n\n参考来源：{', '.join(set(sources))}"
        return {
            **state,
            "generation": generation_with_sources
        }
```

## 常见问题

**Q: Agentic RAG 比普通 RAG 慢吗？**

A: 可能会稍慢一些，因为增加了判断和评估步骤。但通过智能决策，可以避免不必要的检索，整体效率可能更高。

**Q: 如何控制重写次数？**

A: 在 `grade_documents` 函数中添加检索次数限制，避免无限循环。

**Q: 可以使用其他 LLM 吗？**

A: 可以。LangChain 支持多种 LLM，只需替换 `ChatTongyi` 为对应的类，如 `ChatOpenAI`（OpenAI）、`ChatZhipuAI`（智谱AI）等。

**Q: 如何提高检索质量？**

A:

1. 优化分块策略，确保文档块大小和重叠合适
2. 使用高质量的 embedding 模型（如 `text-embedding-v4`）
3. 调整检索数量（`k` 值）
4. 优化问题重写的提示词

**Q: 如何处理多轮对话？**

A: 在状态中添加 `chat_history` 字段，在生成答案时考虑历史对话上下文。

## 总结

通过本教程，您已经学会了如何构建一个 Agentic RAG 系统。核心思路是：

1. **xParse SDK 负责文档解析**：使用 [xParse SDK](/xparse/v1/sdk-python) 将文档转换为 Markdown，再通过 LangChain 分块和向量化构建知识库
2. **LangGraph 负责工作流**：构建智能的检索和生成流程
3. **智能决策**：通过判断、评估、重写等步骤，逐步优化答案质量

相比普通 RAG，Agentic RAG 能够：

* 智能判断是否需要检索
* 评估检索结果的相关性
* 自动重写问题以优化检索
* 生成更准确、更相关的答案

这样，您就有了一个"会思考"的 RAG 系统！

## 下一步

* **查看 [RAG 教程](/xparse/v1/tutorials/rag-tutorial)**：了解基础 RAG 的实现
* **查看 [Agent 教程](/xparse/v1/tutorials/agent-tutorial)**：了解如何使用 LangChain Agent
* **阅读 [xParse SDK 文档](/xparse/v1/quickstart)**：了解 xParse SDK 的详细使用方式