
单个智能体通常可以在单个领域内使用少量工具有效地操作，但即使使用像GPT-4这样强大的模型，它在使用许多工具时也可能效果不佳。

处理复杂任务的一种方法是通过“分而治之”的方式：为每个任务或领域创建一个专门的智能体，并将任务路由到正确的“专家”。这是多智能体网络架构的一个示例。

本笔记本（灵感来自Wu等人的论文《AutoGen: 通过多智能体对话实现下一代LLM应用》）展示了如何使用LangGraph实现这一点。

生成的图将类似于以下图表：

多智能体图

在我们开始之前，快速说明一下：本笔记本和其他多智能体笔记本旨在展示如何在LangGraph中实现某些设计模式。如果该模式符合您的需求，我们建议将其与文档中描述的其他基本模式结合使用，以获得最佳性能。


首先，让我们安装所需的包并设置API密钥：

```python
%%capture --no-stderr
%pip install -U langchain_community langchain_anthropic langchain_experimental matplotlib langgraph

import getpass
import os

def _set_if_undefined(var: str):
    if not os.environ.get(var):
        os.environ[var] = getpass.getpass(f"Please provide your {var}")

_set_if_undefined("ANTHROPIC_API_KEY")
_set_if_undefined("TAVILY_API_KEY")
```

为LangGraph开发设置LangSmith

注册LangSmith以快速发现问题并提高LangGraph项目的性能。LangSmith允许您使用跟踪数据来调试、测试和监控使用LangGraph构建的LLM应用程序——阅读更多关于如何开始的信息。


我们还将定义一些工具，供我们的智能体将来使用：

```python
from typing import Annotated
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.tools import tool
from langchain_experimental.utilities import PythonREPL

tavily_tool = TavilySearchResults(max_results=5)

# 警告：此代码在本地执行，未沙盒化时可能不安全
repl = PythonREPL()

@tool
def python_repl_tool(
    code: Annotated[str, "The python code to execute to generate your chart."],
):
    """使用此工具执行Python代码。如果你想查看某个值的输出，你应该使用`print(...)`打印出来。用户可以看到此输出。"""
    try:
        result = repl.run(code)
    except BaseException as e:
        return f"Failed to execute. Error: {repr(e)}"
    result_str = f"Successfully executed:\n\`\`\`python\n{code}\n\`\`\`\nStdout: {result}"
    return (
        result_str + "\n\nIf you have completed all tasks, respond with FINAL ANSWER."
    )
```


现在我们已经定义了工具并创建了一些辅助函数，我们将创建各个智能体，并使用LangGraph告诉它们如何相互通信。


我们现在需要定义节点。

首先，我们将创建一个实用程序来为每个智能体生成系统提示。

```python
def make_system_prompt(suffix: str) -> str:
    return (
        "You are a helpful AI assistant, collaborating with other assistants."
        " Use the provided tools to progress towards answering the question."
        " If you are unable to fully answer, that's OK, another assistant with different tools "
        " will help where you left off. Execute what you can to make progress."
        " If you or any of the other assistants have the final answer or deliverable,"
        " prefix your response with FINAL ANSWER so the team knows to stop."
        f"\n{suffix}"
    )

from typing import Literal
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_anthropic import ChatAnthropic
from langgraph.prebuilt import create_react_agent
from langgraph.graph import MessagesState, END
from langgraph.types import Command

llm = ChatAnthropic(model="claude-3-5-sonnet-latest")

def get_next_node(last_message: BaseMessage, goto: str):
    if "FINAL ANSWER" in last_message.content:
        # 任何智能体决定工作已完成
        return END
    return goto

# 研究智能体和节点
research_agent = create_react_agent(
    llm,
    tools=[tavily_tool],
    prompt=make_system_prompt(
        "You can only do research. You are working with a chart generator colleague."
    ),
)

def research_node(
    state: MessagesState,
) -> Command[Literal["chart_generator", END]]:
    result = research_agent.invoke(state)
    goto = get_next_node(result["messages"][-1], "chart_generator")
    # 包装在人类消息中，因为并非所有提供程序都允许
    # 在输入消息列表的最后位置使用AI消息
    result["messages"][-1] = HumanMessage(
        content=result["messages"][-1].content, name="researcher"
    )
    return Command(
        update={
            # 与其他智能体共享研究智能体的内部消息历史
            "messages": result["messages"],
        },
        goto=goto,
    )

# 图表生成智能体和节点
# 注意：此操作执行任意代码执行，未沙盒化时可能不安全
chart_agent = create_react_agent(
    llm,
    [python_repl_tool],
    prompt=make_system_prompt(
        "You can only generate charts. You are working with a researcher colleague."
    ),
)

def chart_node(state: MessagesState) -> Command[Literal["researcher", END]]:
    result = chart_agent.invoke(state)
    goto = get_next_node(result["messages"][-1], "researcher")
    # 包装在人类消息中，因为并非所有提供程序都允许
    # 在输入消息列表的最后位置使用AI消息
    result["messages"][-1] = HumanMessage(
        content=result["messages"][-1].content, name="chart_generator"
    )
    return Command(
        update={
            # 与其他智能体共享图表生成智能体的内部消息历史
            "messages": result["messages"],
        },
        goto=goto,
    )
```


我们现在可以将所有内容放在一起并定义图！

```python
from langgraph.graph import StateGraph, START

workflow = StateGraph(MessagesState)
workflow.add_node("researcher", research_node)
workflow.add_node("chart_generator", chart_node)

workflow.add_edge(START, "researcher")
graph = workflow.compile()
```


图创建完成后，您可以调用它！让我们用它来绘制一些统计数据。

```python
events = graph.stream(
    {
        "messages": [
            (
                "user",
                "First, get the UK's GDP over the past 5 years, then make a line chart of it. "
                "Once you make the chart, finish.",
            )
        ],
    },
    # 图中执行的最大步骤数
    {"recursion_limit": 150},
)
for s in events:
    print(s)
    print("----")
```

输出结果将显示智能体之间的交互过程，最终生成一个包含英国过去5年GDP数据的折线图。

多Agent

### 使用本地LLM实现Self-RAG

Self-RAG是一种结合了自我反思/自我评分的RAG（检索增强生成）策略。在论文中，Self-RAG通过以下几个决策点来优化检索和生成过程：

1. **是否从检索器中检索（R）**  
   - 输入：问题（x）或问题（x）和生成结果（y）  
   - 决定何时从检索器R中检索D个文档块  
   - 输出：是、否、继续

2. **检索到的文档块D是否与问题x相关**  
   - 输入：问题（x）和文档块（d）  
   - 判断d是否为解决x提供了有用的信息  
   - 输出：相关、不相关

3. **LLM生成的每个文档块D的生成结果是否与文档块相关（是否存在幻觉等）**  
   - 输入：问题（x）、文档块（d）、生成结果（y）  
   - 判断y中的所有可验证陈述是否由d支持  
   - 输出：完全支持、部分支持、不支持

4. **LLM生成的每个文档块D的生成结果是否对问题x有用**  
   - 输入：问题（x）、生成结果（y）  
   - 判断y是否对解决x有用  
   - 输出：5、4、3、2、1

我们将使用LangGraph从头实现这些思想。

---

### 设置

首先，安装所需的包并设置API密钥。

```python
%%capture --no-stderr
%pip install -U langchain-nomic langchain_community tiktoken langchainhub chromadb langchain langgraph nomic[local]

import getpass
import os

def _set_env(key: str):
    if key not in os.environ:
        os.environ[key] = getpass.getpass(f"{key}:")

_set_env("NOMIC_API_KEY")
```

---

### 本地LLM和嵌入

1. **本地嵌入**  
   使用Nomic的`GPT4AllEmbeddings()`，支持Nomic发布的v1和v1.5嵌入模型。  
   参考[文档](https://docs.nomic.ai/)。

2. **本地LLM**  
   - 下载Ollama应用。  
   - 从[Mistral版本](https://mistral.ai/models/)或[Mixtral版本](https://mixtral.ai/models/)下载模型。  
   - 使用以下命令拉取模型：  
     ```bash
     ollama pull mistral
     ```  
   - 设置本地LLM名称：  
     ```python
     local_llm = "mistral"
     ```

---

### 创建索引

索引3篇博客文章。

```python
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import WebBaseLoader
from langchain_community.vectorstores import Chroma
from langchain_nomic.embeddings import NomicEmbeddings

urls = [
    "https://lilianweng.github.io/posts/2023-06-23-agent/",
    "https://lilianweng.github.io/posts/2023-03-15-prompt-engineering/",
    "https://lilianweng.github.io/posts/2023-10-25-adv-attack-llm/",
]

docs = [WebBaseLoader(url).load() for url in urls]
docs_list = [item for sublist in docs for item in sublist]

text_splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
    chunk_size=250, chunk_overlap=0
)
doc_splits = text_splitter.split_documents(docs_list)

# 添加到向量数据库
vectorstore = Chroma.from_documents(
    documents=doc_splits,
    collection_name="rag-chroma",
    embedding=NomicEmbeddings(model="nomic-embed-text-v1.5", inference_mode="local"),
)
retriever = vectorstore.as_retriever()
```

---

### 检索评分器

```python
from langchain.prompts import PromptTemplate
from langchain_community.chat_models import ChatOllama
from langchain_core.output_parsers import JsonOutputParser

# LLM
llm = ChatOllama(model=local_llm, format="json", temperature=0)

prompt = PromptTemplate(
    template="""You are a grader assessing relevance of a retrieved document to a user question. \n 
    Here is the retrieved document: \n\n {document} \n\n
    Here is the user question: {question} \n
    If the document contains keywords related to the user question, grade it as relevant. \n
    It does not need to be a stringent test. The goal is to filter out erroneous retrievals. \n
    Give a binary score 'yes' or 'no' score to indicate whether the document is relevant to the question. \n
    Provide the binary score as a JSON with a single key 'score' and no premable or explanation.""",
    input_variables=["question", "document"],
)

retrieval_grader = prompt | llm | JsonOutputParser()
question = "agent memory"
docs = retriever.invoke(question)
doc_txt = docs[1].page_content
print(retrieval_grader.invoke({"question": question, "document": doc_txt}))
```

---

### 生成

```python
from langchain import hub
from langchain_core.output_parsers import StrOutputParser

# Prompt
prompt = hub.pull("rlm/rag-prompt")

# LLM
llm = ChatOllama(model=local_llm, temperature=0)

# 后处理
def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

# Chain
rag_chain = prompt | llm | StrOutputParser()

# 运行
generation = rag_chain.invoke({"context": docs, "question": question})
print(generation)
```

---

### 幻觉评分器

```python
# LLM
llm = ChatOllama(model=local_llm, format="json", temperature=0)

# Prompt
prompt = PromptTemplate(
    template="""You are a grader assessing whether an answer is grounded in / supported by a set of facts. \n 
    Here are the facts:
    \n ------- \n
    {documents} 
    \n ------- \n
    Here is the answer: {generation}
    Give a binary score 'yes' or 'no' score to indicate whether the answer is grounded in / supported by a set of facts. \n
    Provide the binary score as a JSON with a single key 'score' and no preamble or explanation.""",
    input_variables=["generation", "documents"],
)

hallucination_grader = prompt | llm | JsonOutputParser()
hallucination_grader.invoke({"documents": docs, "generation": generation})
```

---

### 答案评分器

```python
# LLM
llm = ChatOllama(model=local_llm, format="json", temperature=0)

# Prompt
prompt = PromptTemplate(
    template="""You are a grader assessing whether an answer is useful to resolve a question. \n 
    Here is the answer:
    \n ------- \n
    {generation} 
    \n ------- \n
    Here is the question: {question}
    Give a binary score 'yes' or 'no' to indicate whether the answer is useful to resolve a question. \n
    Provide the binary score as a JSON with a single key 'score' and no preamble or explanation.""",
    input_variables=["generation", "question"],
)

answer_grader = prompt | llm | JsonOutputParser()
answer_grader.invoke({"question": question, "generation": generation})
```

---

### 问题重写器

```python
# LLM
llm = ChatOllama(model=local_llm, temperature=0)

# Prompt
re_write_prompt = PromptTemplate(
    template="""You a question re-writer that converts an input question to a better version that is optimized \n 
     for vectorstore retrieval. Look at the initial and formulate an improved question. \n
     Here is the initial question: \n\n {question}. Improved question with no preamble: \n """,
    input_variables=["generation", "question"],
)

question_rewriter = re_write_prompt | llm | StrOutputParser()
question_rewriter.invoke({"question": question})
```

---

### 构建图

将上述流程封装为图。

```python
from langgraph.graph import END, StateGraph, START

workflow = StateGraph(GraphState)

# 定义节点
workflow.add_node("retrieve", retrieve)  # 检索
workflow.add_node("grade_documents", grade_documents)  # 评分文档
workflow.add_node("generate", generate)  # 生成
workflow.add_node("transform_query", transform_query)  # 重写问题

# 构建图
workflow.add_edge(START, "retrieve")
workflow.add_edge("retrieve", "grade_documents")
workflow.add_conditional_edges(
    "grade_documents",
    decide_to_generate,
    {
        "transform_query": "transform_query",
        "generate": "generate",
    },
)
workflow.add_edge("transform_query", "retrieve")
workflow.add_conditional_edges(
    "generate",
    grade_generation_v_documents_and_question,
    {
        "not supported": "generate",
        "useful": END,
        "not useful": "transform_query",
    },
)

# 编译
app = workflow.compile()
```

---

### 运行

```python
from pprint import pprint

# 运行
inputs = {"question": "Explain how the different types of agent memory work?"}
for output in app.stream(inputs):
    for key, value in output.items():
        # 节点
        pprint(f"Node '{key}':")
        # 可选：打印每个节点的完整状态
        # pprint.pprint(value["keys"], indent=2, width=80, depth=None)
    pprint("\n---\n")

# 最终生成结果
pprint(value["generation"])
```

---

### 总结

通过本教程，我们使用本地LLM和LangGraph实现了Self-RAG策略，优化了检索和生成过程。Self-RAG通过自我反思和评分机制，显著提升了生成结果的质量和相关性。

使用本地模型的自我反思检索

### **SQL 数据库交互智能体教程**

在本教程中，我们将逐步构建一个能够回答有关 SQL 数据库问题的智能体。该智能体的主要功能包括：

1. 从数据库中获取可用表  
2. 确定哪些表与问题相关  
3. 获取相关表的 DDL（数据定义语言）  
4. 根据问题和 DDL 信息生成查询  
5. 使用 LLM 检查查询中的常见错误  
6. 执行查询并返回结果  
7. 修正数据库引擎返回的错误，直到查询成功  
8. 根据结果生成响应

以下是实现该智能体的详细步骤。

---

### **1. 环境准备与依赖安装**

首先，安装所需的库：

```bash
pip install -U langgraph langchain_openai langchain_community
```

设置 OpenAI API 密钥：

```python
import getpass
import os

def _set_env(key: str):
    if key not in os.environ:
        os.environ[key] = getpass.getpass(f"{key}:")

_set_env("OPENAI_API_KEY")
```

---

### **2. 配置数据库**

我们将使用 SQLite 数据库，并加载 `chinook` 示例数据库。该数据库代表一个数字媒体商店。

```python
import requests

url = "https://storage.googleapis.com/benchmarks-artifacts/chinook/Chinook.db"
response = requests.get(url)

if response.status_code == 200:
    with open("Chinook.db", "wb") as file:
        file.write(response.content)
    print("文件已下载并保存为 Chinook.db")
else:
    print(f"文件下载失败，状态码：{response.status_code}")
```

使用 `langchain_community` 中的 `SQLDatabase` 包装器与数据库交互：

```python
from langchain_community.utilities import SQLDatabase

db = SQLDatabase.from_uri("sqlite:///Chinook.db")
print(db.dialect)
print(db.get_usable_table_names())
db.run("SELECT * FROM Artist LIMIT 10;")
```

---

### **3. 定义工具**

我们将定义以下工具供智能体使用：

1. **`list_tables_tool`**：从数据库中获取可用表  
2. **`get_schema_tool`**：获取表的 DDL  
3. **`db_query_tool`**：执行查询并返回结果，如果查询失败则返回错误信息  

```python
from langchain_community.agent_toolkits import SQLDatabaseToolkit
from langchain_openai import ChatOpenAI

toolkit = SQLDatabaseToolkit(db=db, llm=ChatOpenAI(model="gpt-4o"))
tools = toolkit.get_tools()

list_tables_tool = next(tool for tool in tools if tool.name == "sql_db_list_tables")
get_schema_tool = next(tool for tool in tools if tool.name == "sql_db_schema")

print(list_tables_tool.invoke(""))
print(get_schema_tool.invoke("Artist"))
```

手动定义 `db_query_tool`：

```python
from langchain_core.tools import tool

@tool
def db_query_tool(query: str) -> str:
    """
    对数据库执行 SQL 查询并返回结果。
    如果查询不正确，则返回错误信息。
    """
    result = db.run_no_throw(query)
    if not result:
        return "错误：查询失败。请重写查询并重试。"
    return result

print(db_query_tool.invoke("SELECT * FROM Artist LIMIT 10;"))
```

---

### **4. 查询检查**

使用 LLM 检查查询中的常见错误：

```python
from langchain_core.prompts import ChatPromptTemplate

query_check_system = """你是一位 SQL 专家，注重细节。
请检查以下 SQLite 查询中的常见错误，包括：
- 在 NULL 值上使用 NOT IN
- 应使用 UNION ALL 时使用了 UNION
- 在独占范围中使用 BETWEEN
- 谓词中的数据类型不匹配
- 正确引用标识符
- 函数的参数数量是否正确
- 转换为正确的数据类型
- 连接时使用正确的列

如果存在上述错误，请重写查询。如果没有错误，则直接返回原始查询。
"""

query_check_prompt = ChatPromptTemplate.from_messages(
    [("system", query_check_system), ("placeholder", "{messages}")]
)
query_check = query_check_prompt | ChatOpenAI(model="gpt-4o", temperature=0).bind_tools(
    [db_query_tool], tool_choice="required"
)

query_check.invoke({"messages": [("user", "SELECT * FROM Artist LIMIT 10;")]})
```

---

### **5. 构建智能体工作流**

定义智能体的工作流，包括以下节点：

1. **`retrieve`**：检索相关表  
2. **`grade_documents`**：评估表的相关性  
3. **`generate`**：生成查询  
4. **`transform_query`**：转换查询  

```python
from langgraph.prebuilt import ToolNode
from langchain_core.messages import HumanMessage

def retrieve(state):
    question = state["question"]
    tables = list_tables_tool.invoke(question)
    return {"tables": tables}

def grade_documents(state):
    tables = state["tables"]
    schema = get_schema_tool.invoke(tables[0])
    return {"schema": schema}

def generate(state):
    schema = state["schema"]
    question = state["question"]
    query = llm.invoke([HumanMessage(content=f"根据以下 DDL 和问题生成 SQL 查询：{schema}\n问题：{question}")])
    return {"query": query}

def transform_query(state):
    query = state["query"]
    rewritten_query = query_check.invoke({"messages": [("user", query)]})
    return {"query": rewritten_query}

# 构建工作流
workflow = Graph()
workflow.add_node("retrieve", retrieve)
workflow.add_node("grade_documents", grade_documents)
workflow.add_node("generate", generate)
workflow.add_node("transform_query", transform_query)

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

app = workflow.compile()

# 执行工作流
question = "2009 年销售额最高的销售代理是谁？"
for output in app.stream({"question": question}):
    for key, value in output.items():
        print(f"节点：{key}, 输出：{value}")
```

---

### **6. 评估智能体**

使用 LangSmith 对智能体进行评估，包括响应、工具调用和轨迹的评估。

```python
from langsmith.evaluation import evaluate

def predict_sql_agent_answer(example: dict):
    msg = {"messages": ("user", example["input"])}
    messages = app.invoke(msg)
    json_str = messages["messages"][-1].tool_calls[0]["args"]
    response = json_str["final_answer"]
    return {"response": response}

def answer_evaluator(run, example) -> dict:
    input_question = example.inputs["input"]
    reference = example.outputs["output"]
    prediction = run.outputs["response"]

    llm = ChatOpenAI(model="gpt-4-turbo", temperature=0)
    answer_grader = grade_prompt_answer_accuracy | llm

    score = answer_grader.invoke(
        {
            "question": input_question,
            "correct_answer": reference,
            "student_answer": prediction,
        }
    )
    score = score["Score"]

    return {"key": "answer_v_reference_score", "score": score}

try:
    experiment_results = evaluate(
        predict_sql_agent_answer,
        data="SQL Agent Response",
        evaluators=[answer_evaluator],
        num_repetitions=3,
        experiment_prefix="sql-agent-multi-step-response-v-reference",
        metadata={"version": "Chinook, gpt-4o multi-step-agent"},
    )
except:
    print("请设置 LangSmith")
```
完整代码
```py
```
---

### **总结**

本教程实现了一个能够与 SQL 数据库交互的智能体，结合了检索、生成、检查和执行查询的功能。你可以根据实际需求调整工作流和工具，以优化智能体的性能。

SQLAgent

### **Self-RAG 教程（基于 LangGraph）**

本教程将指导你使用 **LangGraph** 构建一个 **Self-RAG（检索增强生成）** 系统。该系统结合了自我反思和自我评分机制，对检索到的文档和生成的回答进行评估。以下是实现该系统的详细步骤。

---

### **1. 环境准备与依赖安装**

首先，安装所需的库：

```bash
pip install -U langchain_community tiktoken langchain-openai langchainhub chromadb
```

---

### **2. 检索器设置**

检索器使用 `WebBaseLoader` 加载博客文章，并使用 `Chroma` 向量存储对文档进行索引。文档会被分割成块以便高效检索。

```python
from langchain_community.document_loaders import WebBaseLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

# 加载博客文章
loader = WebBaseLoader("https://lilianweng.github.io/posts/2023-06-23-agent/")
data = loader.load()

# 将文档分割成块
text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=0)
splits = text_splitter.split_documents(data)

# 创建向量存储
vectorstore = Chroma.from_documents(documents=splits, embedding=OpenAIEmbeddings())
retriever = vectorstore.as_retriever()
```

---

### **3. 自我评分与反思机制**

Self-RAG 的核心是 **自我评分** 和 **自我反思**。我们使用 LLM（如 GPT-4）对检索到的文档和生成的回答进行评分。

#### **3.1 文档评分**

```python
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI

# 定义文档评分提示
grade_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个评分助手，负责评估检索到的文档是否与问题相关。"),
    ("human", "问题：{question}\n文档：{document}\n请评分（1-5）：")
])

# 初始化 LLM
llm = ChatOpenAI(model="gpt-4")

def grade_documents(state):
    question = state["question"]
    documents = state["documents"]
    scores = []
    for doc in documents:
        score = llm.invoke(grade_prompt.format(question=question, document=doc))
        scores.append(score)
    return {"scores": scores}
```

#### **3.2 回答反思**

```python
def reflect_response(state):
    generation = state["generation"]
    reflection = llm.invoke([HumanMessage(content=f"请反思以下回答的质量：{generation}")])
    return {"reflection": reflection}
```

---

### **4. 构建 LangGraph 工作流**

使用 LangGraph 将检索、评分、生成和反思模块连接成一个完整的工作流。

```python
from langgraph.graph import Graph

# 定义节点函数
def retrieve(state):
    question = state["question"]
    documents = retriever.get_relevant_documents(question)
    return {"documents": documents}

def generate(state):
    question = state["question"]
    documents = state["documents"]
    generation = llm.invoke([HumanMessage(content=f"问题：{question}\n文档：{documents}")])
    return {"generation": generation}

def transform_query(state):
    question = state["question"]
    rewritten_question = llm.invoke([HumanMessage(content=f"请重写以下问题：{question}")])
    return {"question": rewritten_question}

# 构建工作流
workflow = Graph()
workflow.add_node("retrieve", retrieve)
workflow.add_node("grade_documents", grade_documents)
workflow.add_node("generate", generate)
workflow.add_node("transform_query", transform_query)

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

app = workflow.compile()

# 执行工作流
question = "智能体系统的关键组件是什么？"
for output in app.stream({"question": question}):
    for key, value in output.items():
        print(f"节点：{key}, 输出：{value}")
```

---

### **5. 完整代码**

以下是完整的 Self-RAG 实现代码：

```python
from langchain_community.document_loaders import WebBaseLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.messages import HumanMessage
from langgraph.graph import Graph

# 加载博客文章
loader = WebBaseLoader("https://lilianweng.github.io/posts/2023-06-23-agent/")
data = loader.load()

# 将文档分割成块
text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=0)
splits = text_splitter.split_documents(data)

# 创建向量存储
vectorstore = Chroma.from_documents(documents=splits, embedding=OpenAIEmbeddings())
retriever = vectorstore.as_retriever()

# 初始化 LLM
llm = ChatOpenAI(model="gpt-4")

# 定义文档评分提示
grade_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个评分助手，负责评估检索到的文档是否与问题相关。"),
    ("human", "问题：{question}\n文档：{document}\n请评分（1-5）：")
])

# 定义节点函数
def retrieve(state):
    question = state["question"]
    documents = retriever.get_relevant_documents(question)
    return {"documents": documents}

def grade_documents(state):
    question = state["question"]
    documents = state["documents"]
    scores = []
    for doc in documents:
        score = llm.invoke(grade_prompt.format(question=question, document=doc))
        scores.append(score)
    return {"scores": scores}

def generate(state):
    question = state["question"]
    documents = state["documents"]
    generation = llm.invoke([HumanMessage(content=f"问题：{question}\n文档：{documents}")])
    return {"generation": generation}

def transform_query(state):
    question = state["question"]
    rewritten_question = llm.invoke([HumanMessage(content=f"请重写以下问题：{question}")])
    return {"question": rewritten_question}

# 构建工作流
workflow = Graph()
workflow.add_node("retrieve", retrieve)
workflow.add_node("grade_documents", grade_documents)
workflow.add_node("generate", generate)
workflow.add_node("transform_query", transform_query)

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

app = workflow.compile()

# 执行工作流
question = "智能体系统的关键组件是什么？"
for output in app.stream({"question": question}):
    for key, value in output.items():
        print(f"节点：{key}, 输出：{value}")
```

---

### **总结**

本教程实现了一个 **Self-RAG** 系统，结合了检索、生成、评分和反思功能。你可以根据实际需求调整工作流和提示词，以优化系统性能。

自我反思检索

### 使用LangGraph实现Corrective RAG（自校正检索增强生成）

在本教程中，我们将使用LangGraph框架实现一个Corrective RAG（自校正检索增强生成）系统。该系统能够对检索到的文档进行自校正，并根据文档的相关性动态补充Web搜索，以生成更准确的回答。以下是详细的实现步骤。

---

### 1. 环境设置
首先，安装所需的Python包并设置API密钥。

```python
%%capture --no-stderr
%pip install -U langchain_community tiktoken langchain-openai langchainhub chromadb langchain langgraph tavily-python

import getpass
import os

def _set_env(var: str):
    if not os.environ.get(var):
        os.environ[var] = getpass.getpass(f"{var}: ")

_set_env("OPENAI_API_KEY")
_set_env("TAVILY_API_KEY")
```

---

### 2. 创建索引
我们加载并索引3篇博客文章，以便后续检索。

```python
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import WebBaseLoader
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

urls = [
    "https://lilianweng.github.io/posts/2023-06-23-agent/",
    "https://lilianweng.github.io/posts/2023-03-15-prompt-engineering/",
    "https://lilianweng.github.io/posts/2023-10-25-adv-attack-llm/",
]

docs = [WebBaseLoader(url).load() for url in urls]
docs_list = [item for sublist in docs for item in sublist]

text_splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
    chunk_size=250, chunk_overlap=0
)
doc_splits = text_splitter.split_documents(docs_list)

# 添加到向量数据库
vectorstore = Chroma.from_documents(
    documents=doc_splits,
    collection_name="rag-chroma",
    embedding=OpenAIEmbeddings(),
)
retriever = vectorstore.as_retriever()
```

---

### 3. 定义检索评分器
我们定义一个评分器，用于评估检索到的文档是否与问题相关。

```python
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field

# 数据模型
class GradeDocuments(BaseModel):
    """用于评估检索到的文档是否相关的二元评分。"""
    binary_score: str = Field(description="文档是否与问题相关，'yes' 或 'no'")

# LLM 配置
llm = ChatOpenAI(model="gpt-3.5-turbo-0125", temperature=0)
structured_llm_grader = llm.with_structured_output(GradeDocuments)

# 提示模板
system = """你是一个评估检索到的文档与用户问题相关性的评分器。\n 
    如果文档包含与问题相关的关键词或语义，则评分为相关。\n
    给出一个二元评分 'yes' 或 'no' 来表示文档是否与问题相关。"""
grade_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", system),
        ("human", "检索到的文档: \n\n {document} \n\n 用户问题: {question}"),
    ]
)

retrieval_grader = grade_prompt | structured_llm_grader
```

---

### 4. 定义生成器
我们定义一个生成器，用于根据检索到的文档生成回答。

```python
from langchain import hub
from langchain_core.output_parsers import StrOutputParser

# 提示模板
prompt = hub.pull("rlm/rag-prompt")

# LLM 配置
llm = ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0)

# 后处理函数
def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

# 生成链
rag_chain = prompt | llm | StrOutputParser()
```

---

### 5. 定义问题重写器
我们定义一个重写器，用于优化问题以便进行Web搜索。

```python
# LLM 配置
llm = ChatOpenAI(model="gpt-3.5-turbo-0125", temperature=0)

# 提示模板
system = """你是一个问题重写器，将输入问题转换为更适合Web搜索的版本。\n 
     查看输入并尝试推理其潜在的语义意图/含义。"""
re_write_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", system),
        ("human", "初始问题: \n\n {question} \n 生成一个改进的问题。"),
    ]
)

question_rewriter = re_write_prompt | llm | StrOutputParser()
```

---

### 6. 定义Web搜索工具
我们使用Tavily Search进行Web搜索。

```python
from langchain_community.tools.tavily_search import TavilySearchResults

web_search_tool = TavilySearchResults(k=3)
```

---

### 7. 定义图状态和节点
我们定义图的状态和各个节点。

```python
from typing import List, TypedDict
from langchain.schema import Document

class GraphState(TypedDict):
    """表示图的状态。"""
    question: str
    generation: str
    web_search: str
    documents: List[Document]

def retrieve(state):
    """检索文档。"""
    print("---RETRIEVE---")
    question = state["question"]
    documents = retriever.invoke(question)
    return {"documents": documents, "question": question}

def generate(state):
    """生成回答。"""
    print("---GENERATE---")
    question = state["question"]
    documents = state["documents"]
    generation = rag_chain.invoke({"context": documents, "question": question})
    return {"documents": documents, "question": question, "generation": generation}

def grade_documents(state):
    """评估文档的相关性。"""
    print("---CHECK DOCUMENT RELEVANCE TO QUESTION---")
    question = state["question"]
    documents = state["documents"]
    filtered_docs = []
    web_search = "No"
    for d in documents:
        score = retrieval_grader.invoke({"question": question, "document": d.page_content})
        if score.binary_score == "yes":
            print("---GRADE: DOCUMENT RELEVANT---")
            filtered_docs.append(d)
        else:
            print("---GRADE: DOCUMENT NOT RELEVANT---")
            web_search = "Yes"
    return {"documents": filtered_docs, "question": question, "web_search": web_search}

def transform_query(state):
    """重写问题。"""
    print("---TRANSFORM QUERY---")
    question = state["question"]
    better_question = question_rewriter.invoke({"question": question})
    return {"documents": state["documents"], "question": better_question}

def web_search(state):
    """进行Web搜索。"""
    print("---WEB SEARCH---")
    question = state["question"]
    docs = web_search_tool.invoke({"query": question})
    web_results = Document(page_content="\n".join([d["content"] for d in docs]))
    return {"documents": [web_results], "question": question}

def decide_to_generate(state):
    """决定是否生成回答或重写问题。"""
    print("---ASSESS GRADED DOCUMENTS---")
    if state["web_search"] == "Yes":
        print("---DECISION: ALL DOCUMENTS ARE NOT RELEVANT TO QUESTION, TRANSFORM QUERY---")
        return "transform_query"
    else:
        print("---DECISION: GENERATE---")
        return "generate"
```

---

### 8. 编译并运行图
我们编译图并运行它来生成回答。

```python
from langgraph.graph import END, StateGraph, START

workflow = StateGraph(GraphState)

# 定义节点
workflow.add_node("retrieve", retrieve)
workflow.add_node("grade_documents", grade_documents)
workflow.add_node("generate", generate)
workflow.add_node("transform_query", transform_query)
workflow.add_node("web_search_node", web_search)

# 构建图
workflow.add_edge(START, "retrieve")
workflow.add_edge("retrieve", "grade_documents")
workflow.add_conditional_edges(
    "grade_documents",
    decide_to_generate,
    {
        "transform_query": "transform_query",
        "generate": "generate",
    },
)
workflow.add_edge("transform_query", "web_search_node")
workflow.add_edge("web_search_node", "generate")
workflow.add_edge("generate", END)

# 编译
app = workflow.compile()

# 运行
from pprint import pprint

inputs = {"question": "What are the types of agent memory?"}
for output in app.stream(inputs):
    for key, value in output.items():
        pprint(f"Node '{key}':")
    pprint("\n---\n")

# 最终生成
pprint(value["generation"])
```

---

通过本教程，你可以掌握如何使用LangGraph实现一个Corrective RAG系统，并根据文档的相关性动态补充Web搜索以生成更准确的回答。希望这对你的学习和开发有所帮助！

```py
%%capture --no-stderr
%pip install -U langchain_community tiktoken langchain-openai langchainhub chromadb langchain langgraph tavily-python

import getpass
import os

def _set_env(var: str):
    if not os.environ.get(var):
        os.environ[var] = getpass.getpass(f"{var}: ")

_set_env("OPENAI_API_KEY")
_set_env("TAVILY_API_KEY")

from typing import Dict, TypedDict

from langchain import hub
from langchain.retrievers import TavilySearchAPIRetriever
from langchain_community.document_loaders import WebBaseLoader
from langchain_community.vectorstores import Chroma
from langchain_core.documents import Document
from langchain_core.messages import BaseMessage
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langgraph.graph import END, Graph

# 创建索引
loader = WebBaseLoader("https://lilianweng.github.io/posts/2023-06-23-agent/")
docs = loader.load()

vectorstore = Chroma.from_documents(docs, embedding=OpenAIEmbeddings())
retriever = vectorstore.as_retriever()

# 定义状态
class GraphState(TypedDict):
    question: str
    documents: list[Document]
    generation: str

# 检索节点
def retrieve(state: GraphState):
    print("---RETRIEVE---")
    question = state["question"]
    documents = retriever.invoke(question)
    return {"documents": documents, "question": question}

# 文档评分节点
def grade_documents(state: GraphState):
    print("---GRADE DOCUMENTS---")
    question = state["question"]
    documents = state["documents"]

    # 评分逻辑
    filtered_docs = []
    for d in documents:
        # 假设评分逻辑为：文档内容包含问题关键词
        if any(word in d.page_content.lower() for word in question.lower().split()):
            filtered_docs.append(d)

    return {"documents": filtered_docs, "question": question}

# 决定是否生成或补充Web搜索
def decide_to_generate(state: GraphState):
    print("---DECIDE TO GENERATE---")
    if len(state["documents"]) >= 1:
        return "generate"
    else:
        return "transform_query"

# 转换查询节点
def transform_query(state: GraphState):
    print("---TRANSFORM QUERY---")
    question = state["question"]
    # 假设将问题转换为更具体的查询
    transformed_query = f"{question} (with additional context from web search)"
    return {"question": transformed_query}

# Web搜索节点
def web_search_node(state: GraphState):
    print("---WEB SEARCH---")
    question = state["question"]
    retriever = TavilySearchAPIRetriever(k=3)
    documents = retriever.invoke(question)
    return {"documents": documents, "question": question}

# 生成节点
def generate(state: GraphState):
    print("---GENERATE---")
    question = state["question"]
    documents = state["documents"]

    # 生成逻辑
    prompt = ChatPromptTemplate.from_template("Answer the question based on the context:\n\n{context}\n\nQuestion: {question}")
    model = ChatOpenAI(model="gpt-3.5-turbo")
    chain = prompt | model | StrOutputParser()

    context = "\n".join([d.page_content for d in documents])
    generation = chain.invoke({"context": context, "question": question})
    return {"generation": generation}

# 构建图
workflow = Graph()

workflow.add_node("retrieve", retrieve)
workflow.add_node("grade_documents", grade_documents)
workflow.add_node("transform_query", transform_query)
workflow.add_node("web_search_node", web_search_node)
workflow.add_node("generate", generate)

workflow.add_edge(START, "retrieve")
workflow.add_edge("retrieve", "grade_documents")
workflow.add_conditional_edges(
    "grade_documents",
    decide_to_generate,
    {
        "transform_query": "transform_query",
        "generate": "generate",
    },
)
workflow.add_edge("transform_query", "web_search_node")
workflow.add_edge("web_search_node", "generate")
workflow.add_edge("generate", END)

# 编译
app = workflow.compile()

# 运行
from pprint import pprint

inputs = {"question": "What are the types of agent memory?"}
for output in app.stream(inputs):
    for key, value in output.items():
        pprint(f"Node '{key}':")
    pprint("\n---\n")

# 最终生成
pprint(value["generation"])

```完整代码