自适应 RAG¶

自适应 RAG 是一种 RAG 策略，它将 (1) 查询分析 与 (2) 主动/自校正 RAG 结合在一起。

在 论文 中，他们报告了用于跨以下路线的查询分析：

• 无检索
• 单次 RAG
• 迭代式 RAG

让我们使用 LangGraph 在此基础上进行构建。

在我们的实现中，我们将进行以下路线：

• 网络搜索：用于与近期事件相关的查询
• 自校正 RAG：用于与我们的索引相关的查询

设置¶

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

在 [ ]

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

%%capture --no-stderr ! pip install -U langchain_community tiktoken langchain-openai langchain-cohere 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("COHERE_API_KEY")
_set_env("TAVILY_API_KEY")

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("COHERE_API_KEY") _set_env("TAVILY_API_KEY")

为 LangGraph 开发设置 LangSmith

注册 LangSmith 以快速发现问题并提高 LangGraph 项目的性能。LangSmith 允许您使用跟踪数据来调试、测试和监控使用 LangGraph 构建的 LLM 应用程序 - 了解如何开始使用 此处。

创建索引¶

在 [1]

### Build Index

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

### from langchain_cohere import CohereEmbeddings

# Set embeddings
embd = OpenAIEmbeddings()

# Docs to index
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/",
]

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

# Split
text_splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
    chunk_size=500, chunk_overlap=0
)
doc_splits = text_splitter.split_documents(docs_list)

# Add to vectorstore
vectorstore = Chroma.from_documents(
    documents=doc_splits,
    collection_name="rag-chroma",
    embedding=embd,
)
retriever = vectorstore.as_retriever()

### 从索引构建 from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_community.document_loaders import WebBaseLoader from langchain_community.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings ### from langchain_cohere import CohereEmbeddings # 设置嵌入 embd = 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=500, chunk_overlap=0 ) doc_splits = text_splitter.split_documents(docs_list) # 添加到向量存储 vectorstore = Chroma.from_documents( documents=doc_splits, collection_name="rag-chroma", embedding=embd, ) retriever = vectorstore.as_retriever()

LLMs¶

使用 LangChain 与 Pydantic

此笔记本使用 Pydantic v2 BaseModel，它需要 langchain-core >= 0.3。使用 langchain-core < 0.3 将导致由于混合了 Pydantic v1 和 v2 BaseModels 而导致的错误。

在 [3]

### Router

from typing import Literal

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

from pydantic import BaseModel, Field


# Data model
class RouteQuery(BaseModel):
    """Route a user query to the most relevant datasource."""

    datasource: Literal["vectorstore", "web_search"] = Field(
        ...,
        description="Given a user question choose to route it to web search or a vectorstore.",
    )


# LLM with function call
llm = ChatOpenAI(model="gpt-3.5-turbo-0125", temperature=0)
structured_llm_router = llm.with_structured_output(RouteQuery)

# Prompt
system = """You are an expert at routing a user question to a vectorstore or web search.
The vectorstore contains documents related to agents, prompt engineering, and adversarial attacks.
Use the vectorstore for questions on these topics. Otherwise, use web-search."""
route_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", system),
        ("human", "{question}"),
    ]
)

question_router = route_prompt | structured_llm_router
print(
    question_router.invoke(
        {"question": "Who will the Bears draft first in the NFL draft?"}
    )
)
print(question_router.invoke({"question": "What are the types of agent memory?"}))

### 路由器 from typing import Literal from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from pydantic import BaseModel, Field # 数据模型 class RouteQuery(BaseModel): """将用户查询路由到最相关的来源。""" datasource: Literal["vectorstore", "web_search"] = Field( ..., description="给定用户问题，选择将其路由到网络搜索还是向量存储。", ) # 带有函数调用的 LLM llm = ChatOpenAI(model="gpt-3.5-turbo-0125", temperature=0) structured_llm_router = llm.with_structured_output(RouteQuery) # 提示系统 = """您是将用户问题路由到向量存储或网络搜索的专家。向量存储包含与代理、提示工程和对抗攻击相关的文档。将有关这些主题的问题使用向量存储。否则，请使用网络搜索。""" route_prompt = ChatPromptTemplate.from_messages( [ ("system", system), ("human", "{question}"), ] ) question_router = route_prompt | structured_llm_router print( question_router.invoke( {"question": "熊队将在 NFL 选秀中首先选谁？"} ) ) print(question_router.invoke({"question": "代理内存有哪些类型？"}))

datasource='web_search'
datasource='vectorstore'

在 [4]

### Retrieval Grader


# Data model
class GradeDocuments(BaseModel):
    """Binary score for relevance check on retrieved documents."""

    binary_score: str = Field(
        description="Documents are relevant to the question, 'yes' or 'no'"
    )


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

# Prompt
system = """You are a grader assessing relevance of a retrieved document to a user question. \n 
    If the document contains keyword(s) or semantic meaning 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."""
grade_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", system),
        ("human", "Retrieved document: \n\n {document} \n\n User question: {question}"),
    ]
)

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

### 检索评分器 # 数据模型 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) # 提示系统 = """您是评估检索到的文档与用户问题相关性的评分员。\n 如果文档包含与用户问题相关的关键词或语义意义，则将其评分为相关。\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 question = "代理内存" docs = retriever.invoke(question) doc_txt = docs[1].page_content print(retrieval_grader.invoke({"question": question, "document": doc_txt}))

binary_score='no'

在 [5]

### Generate

from langchain import hub
from langchain_core.output_parsers import StrOutputParser

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

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


# Post-processing
def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)


# Chain
rag_chain = prompt | llm | StrOutputParser()

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

### 生成 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() # 运行生成 generation = rag_chain.invoke({"context": docs, "question": question}) print(generation)

The design of generative agents combines LLM with memory, planning, and reflection mechanisms to enable agents to behave based on past experience and interact with other agents. Memory stream is a long-term memory module that records agents' experiences in natural language. The retrieval model surfaces context to inform the agent's behavior based on relevance, recency, and importance.

在 [6]

### Hallucination Grader


# Data model
class GradeHallucinations(BaseModel):
    """Binary score for hallucination present in generation answer."""

    binary_score: str = Field(
        description="Answer is grounded in the facts, 'yes' or 'no'"
    )


# LLM with function call
llm = ChatOpenAI(model="gpt-3.5-turbo-0125", temperature=0)
structured_llm_grader = llm.with_structured_output(GradeHallucinations)

# Prompt
system = """You are a grader assessing whether an LLM generation is grounded in / supported by a set of retrieved facts. \n 
     Give a binary score 'yes' or 'no'. 'Yes' means that the answer is grounded in / supported by the set of facts."""
hallucination_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", system),
        ("human", "Set of facts: \n\n {documents} \n\n LLM generation: {generation}"),
    ]
)

hallucination_grader = hallucination_prompt | structured_llm_grader
hallucination_grader.invoke({"documents": docs, "generation": generation})

### 幻觉评分器 # 数据模型 class GradeHallucinations(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(GradeHallucinations) # 提示系统 = """您是评估 LLM 生成是否以一组检索到的事实为基础/支持的评分员。\n 给出一个二进制分数 'yes' 或 'no'。'Yes' 表示答案以一组事实为基础/支持。""" hallucination_prompt = ChatPromptTemplate.from_messages( [ ("system", system), ("human", "一组事实：\n\n {documents} \n\n LLM 生成：{generation}"), ] ) hallucination_grader = hallucination_prompt | structured_llm_grader hallucination_grader.invoke({"documents": docs, "generation": generation})

Out[6]

GradeHallucinations(binary_score='yes')

在 [7]

### Answer Grader


# Data model
class GradeAnswer(BaseModel):
    """Binary score to assess answer addresses question."""

    binary_score: str = Field(
        description="Answer addresses the question, 'yes' or 'no'"
    )


# LLM with function call
llm = ChatOpenAI(model="gpt-3.5-turbo-0125", temperature=0)
structured_llm_grader = llm.with_structured_output(GradeAnswer)

# Prompt
system = """You are a grader assessing whether an answer addresses / resolves a question \n 
     Give a binary score 'yes' or 'no'. Yes' means that the answer resolves the question."""
answer_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", system),
        ("human", "User question: \n\n {question} \n\n LLM generation: {generation}"),
    ]
)

answer_grader = answer_prompt | structured_llm_grader
answer_grader.invoke({"question": question, "generation": generation})

### 答案评分器 # 数据模型 class GradeAnswer(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(GradeAnswer) # 提示系统 = """您是评估答案是否解决问题的评分员\n 给出一个二进制分数 'yes' 或 'no'。'Yes' 表示答案解决了问题。""" answer_prompt = ChatPromptTemplate.from_messages( [ ("system", system), ("human", "用户问题：\n\n {question} \n\n LLM 生成：{generation}"), ] ) answer_grader = answer_prompt | structured_llm_grader answer_grader.invoke({"question": question, "generation": generation})

Out[7]

GradeAnswer(binary_score='yes')

在 [8]

### Question Re-writer

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

# Prompt
system = """You a question re-writer that converts an input question to a better version that is optimized \n 
     for vectorstore retrieval. Look at the input and try to reason about the underlying semantic intent / meaning."""
re_write_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", system),
        (
            "human",
            "Here is the initial question: \n\n {question} \n Formulate an improved question.",
        ),
    ]
)

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

### 问题重写器 # LLM llm = ChatOpenAI(model="gpt-3.5-turbo-0125", temperature=0) # 提示系统 = """您是一个问题重写器，它将输入问题转换为一个优化\n 用于向量存储检索的版本。查看输入并尝试推理底层语义意图/含义。""" re_write_prompt = ChatPromptTemplate.from_messages( [ ("system", system), ( "human", "这是初始问题：\n\n {question} \n 重新措辞一个更好的问题。", ), ] ) question_rewriter = re_write_prompt | llm | StrOutputParser() question_rewriter.invoke({"question": question})

Out[8]

"What is the role of memory in an agent's functioning?"

网络搜索工具¶

在 [9]

### Search

from langchain_community.tools.tavily_search import TavilySearchResults

web_search_tool = TavilySearchResults(k=3)

### 搜索 from langchain_community.tools.tavily_search import TavilySearchResults web_search_tool = TavilySearchResults(k=3)

构建图¶

将流程捕获为图。

定义图状态¶

在 [10]

from typing import List

from typing_extensions import TypedDict


class GraphState(TypedDict):
    """
    Represents the state of our graph.

    Attributes:
        question: question
        generation: LLM generation
        documents: list of documents
    """

    question: str
    generation: str
    documents: List[str]

from typing import List from typing_extensions import TypedDict class GraphState(TypedDict): """ 表示图的状态。 属性： question: question generation: LLM 生成 documents: 文档列表 """ question: str generation: str documents: List[str]

定义图流程¶

在 [15]

from langchain.schema import Document


def retrieve(state):
    """
    Retrieve documents

    Args:
        state (dict): The current graph state

    Returns:
        state (dict): New key added to state, documents, that contains retrieved documents
    """
    print("---RETRIEVE---")
    question = state["question"]

    # Retrieval
    documents = retriever.invoke(question)
    return {"documents": documents, "question": question}


def generate(state):
    """
    Generate answer

    Args:
        state (dict): The current graph state

    Returns:
        state (dict): New key added to state, generation, that contains LLM generation
    """
    print("---GENERATE---")
    question = state["question"]
    documents = state["documents"]

    # RAG generation
    generation = rag_chain.invoke({"context": documents, "question": question})
    return {"documents": documents, "question": question, "generation": generation}


def grade_documents(state):
    """
    Determines whether the retrieved documents are relevant to the question.

    Args:
        state (dict): The current graph state

    Returns:
        state (dict): Updates documents key with only filtered relevant documents
    """

    print("---CHECK DOCUMENT RELEVANCE TO QUESTION---")
    question = state["question"]
    documents = state["documents"]

    # Score each doc
    filtered_docs = []
    for d in documents:
        score = retrieval_grader.invoke(
            {"question": question, "document": d.page_content}
        )
        grade = score.binary_score
        if grade == "yes":
            print("---GRADE: DOCUMENT RELEVANT---")
            filtered_docs.append(d)
        else:
            print("---GRADE: DOCUMENT NOT RELEVANT---")
            continue
    return {"documents": filtered_docs, "question": question}


def transform_query(state):
    """
    Transform the query to produce a better question.

    Args:
        state (dict): The current graph state

    Returns:
        state (dict): Updates question key with a re-phrased question
    """

    print("---TRANSFORM QUERY---")
    question = state["question"]
    documents = state["documents"]

    # Re-write question
    better_question = question_rewriter.invoke({"question": question})
    return {"documents": documents, "question": better_question}


def web_search(state):
    """
    Web search based on the re-phrased question.

    Args:
        state (dict): The current graph state

    Returns:
        state (dict): Updates documents key with appended web results
    """

    print("---WEB SEARCH---")
    question = state["question"]

    # Web search
    docs = web_search_tool.invoke({"query": question})
    web_results = "\n".join([d["content"] for d in docs])
    web_results = Document(page_content=web_results)

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


### Edges ###


def route_question(state):
    """
    Route question to web search or RAG.

    Args:
        state (dict): The current graph state

    Returns:
        str: Next node to call
    """

    print("---ROUTE QUESTION---")
    question = state["question"]
    source = question_router.invoke({"question": question})
    if source.datasource == "web_search":
        print("---ROUTE QUESTION TO WEB SEARCH---")
        return "web_search"
    elif source.datasource == "vectorstore":
        print("---ROUTE QUESTION TO RAG---")
        return "vectorstore"


def decide_to_generate(state):
    """
    Determines whether to generate an answer, or re-generate a question.

    Args:
        state (dict): The current graph state

    Returns:
        str: Binary decision for next node to call
    """

    print("---ASSESS GRADED DOCUMENTS---")
    state["question"]
    filtered_documents = state["documents"]

    if not filtered_documents:
        # All documents have been filtered check_relevance
        # We will re-generate a new query
        print(
            "---DECISION: ALL DOCUMENTS ARE NOT RELEVANT TO QUESTION, TRANSFORM QUERY---"
        )
        return "transform_query"
    else:
        # We have relevant documents, so generate answer
        print("---DECISION: GENERATE---")
        return "generate"


def grade_generation_v_documents_and_question(state):
    """
    Determines whether the generation is grounded in the document and answers question.

    Args:
        state (dict): The current graph state

    Returns:
        str: Decision for next node to call
    """

    print("---CHECK HALLUCINATIONS---")
    question = state["question"]
    documents = state["documents"]
    generation = state["generation"]

    score = hallucination_grader.invoke(
        {"documents": documents, "generation": generation}
    )
    grade = score.binary_score

    # Check hallucination
    if grade == "yes":
        print("---DECISION: GENERATION IS GROUNDED IN DOCUMENTS---")
        # Check question-answering
        print("---GRADE GENERATION vs QUESTION---")
        score = answer_grader.invoke({"question": question, "generation": generation})
        grade = score.binary_score
        if grade == "yes":
            print("---DECISION: GENERATION ADDRESSES QUESTION---")
            return "useful"
        else:
            print("---DECISION: GENERATION DOES NOT ADDRESS QUESTION---")
            return "not useful"
    else:
        pprint("---DECISION: GENERATION IS NOT GROUNDED IN DOCUMENTS, RE-TRY---")
        return "not supported"

从 langchain.schema 导入 Document def retrieve(state): """ 检索文档 Args: state (dict): 当前图状态 Returns: state (dict): 添加到 state 的新键，documents，包含检索到的文档 """ 打印("---RETRIEVE---") question = state["question"] # 检索 documents = retriever.invoke(question) 返回 {"documents": documents, "question": question} def generate(state): """ 生成答案 Args: state (dict): 当前图状态 Returns: state (dict): 添加到 state 的新键，generation，包含 LLM 生成 """ 打印("---GENERATE---") question = state["question"] documents = state["documents"] # RAG 生成 generation = rag_chain.invoke({"context": documents, "question": question}) 返回 {"documents": documents, "question": question, "generation": generation} def grade_documents(state): """ 判断检索到的文档是否与问题相关。 Args: state (dict): 当前图状态 Returns: state (dict): 使用仅过滤后的相关文档更新 documents 键 """ 打印("---CHECK DOCUMENT RELEVANCE TO QUESTION---") question = state["question"] documents = state["documents"] # 对每个文档进行评分 filtered_docs = [] for d in documents: score = retrieval_grader.invoke( {"question": question, "document": d.page_content} ) grade = score.binary_score if grade == "yes": 打印("---GRADE: DOCUMENT RELEVANT---") filtered_docs.append(d) 否则: 打印("---GRADE: DOCUMENT NOT RELEVANT---") 继续 返回 {"documents": filtered_docs, "question": question} def transform_query(state): """ 转换查询以生成更好的问题。 Args: state (dict): 当前图状态 Returns: state (dict): 使用重新措辞的问题更新 question 键 """ 打印("---TRANSFORM QUERY---") question = state["question"] documents = state["documents"] # 重新措辞问题 better_question = question_rewriter.invoke({"question": question}) 返回 {"documents": documents, "question": better_question} def web_search(state): """ 基于重新措辞的问题进行网络搜索。 Args: state (dict): 当前图状态 Returns: state (dict): 使用追加的网络搜索结果更新 documents 键 """ 打印("---WEB SEARCH---") question = state["question"] # 网络搜索 docs = web_search_tool.invoke({"query": question}) web_results = "\n".join([d["content"] for d in docs]) web_results = Document(page_content=web_results) 返回 {"documents": web_results, "question": question} ### 边缘 ### def route_question(state): """ 将问题路由到网络搜索或 RAG。 Args: state (dict): 当前图状态 Returns: str: 要调用的下一个节点 """ 打印("---ROUTE QUESTION---") question = state["question"] source = question_router.invoke({"question": question}) if source.datasource == "web_search": 打印("---ROUTE QUESTION TO WEB SEARCH---") 返回 "web_search" 否则 if source.datasource == "vectorstore": 打印("---ROUTE QUESTION TO RAG---") 返回 "vectorstore" def decide_to_generate(state): """ 判断是生成答案还是重新生成问题。 Args: state (dict): 当前图状态 Returns: str: 要调用的下一个节点的二进制决策 """ 打印("---ASSESS GRADED DOCUMENTS---") state["question"] filtered_documents = state["documents"] if not filtered_documents: # 所有文档都被过滤 check_relevance # 我们将重新生成一个新的查询 打印( "---DECISION: ALL DOCUMENTS ARE NOT RELEVANT TO QUESTION, TRANSFORM QUERY---" ) 返回 "transform_query" 否则: # 我们有相关的文档，所以生成答案 打印("---DECISION: GENERATE---") 返回 "generate" def grade_generation_v_documents_and_question(state): """ 判断生成是否基于文档并回答问题。 Args: state (dict): 当前图状态 Returns: str: 要调用的下一个节点的决策 """ 打印("---CHECK HALLUCINATIONS---") question = state["question"] documents = state["documents"] generation = state["generation"] score = hallucination_grader.invoke( {"documents": documents, "generation": generation} ) grade = score.binary_score # 检查幻觉 if grade == "yes": 打印("---DECISION: GENERATION IS GROUNDED IN DOCUMENTS---") # 检查问答 打印("---GRADE GENERATION vs QUESTION---") score = answer_grader.invoke({"question": question, "generation": generation}) grade = score.binary_score if grade == "yes": 打印("---DECISION: GENERATION ADDRESSES QUESTION---") 返回 "useful" 否则: 打印("---DECISION: GENERATION DOES NOT ADDRESS QUESTION---") 返回 "not useful" 否则: pprint("---DECISION: GENERATION IS NOT GROUNDED IN DOCUMENTS, RE-TRY---") 返回 "not supported"

编译图¶

在 [16]

from langgraph.graph import END, StateGraph, START

workflow = StateGraph(GraphState)

# Define the nodes
workflow.add_node("web_search", web_search)  # web search
workflow.add_node("retrieve", retrieve)  # retrieve
workflow.add_node("grade_documents", grade_documents)  # grade documents
workflow.add_node("generate", generate)  # generatae
workflow.add_node("transform_query", transform_query)  # transform_query

# Build graph
workflow.add_conditional_edges(
    START,
    route_question,
    {
        "web_search": "web_search",
        "vectorstore": "retrieve",
    },
)
workflow.add_edge("web_search", "generate")
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",
    },
)

# Compile
app = workflow.compile()

从 langgraph.graph 导入 END, StateGraph, START workflow = StateGraph(GraphState) # 定义节点 workflow.add_node("web_search", web_search) # 网络搜索 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_conditional_edges( START, route_question, { "web_search": "web_search", "vectorstore": "retrieve", }, ) workflow.add_edge("web_search", "generate") 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()

使用图¶

在 [17]

from pprint import pprint

# Run
inputs = {
    "question": "What player at the Bears expected to draft first in the 2024 NFL draft?"
}
for output in app.stream(inputs):
    for key, value in output.items():
        # Node
        pprint(f"Node '{key}':")
        # Optional: print full state at each node
        # pprint.pprint(value["keys"], indent=2, width=80, depth=None)
    pprint("\n---\n")

# Final generation
pprint(value["generation"])

从 pprint 导入 pprint # 运行 inputs = { "question": "2024 年 NFL 选秀中，熊队预计首轮会选什么球员？" } for output in app.stream(inputs): for key, value in output.items(): # 节点 pprint(f"节点 '{key}':") # 可选：在每个节点打印完整状态 # pprint.pprint(value["keys"], indent=2, width=80, depth=None) pprint("\n---\n") # 最终生成 pprint(value["generation"])

---ROUTE QUESTION---
---ROUTE QUESTION TO WEB SEARCH---
---WEB SEARCH---
"Node 'web_search':"
'\n---\n'
---GENERATE---
---CHECK HALLUCINATIONS---
---DECISION: GENERATION IS GROUNDED IN DOCUMENTS---
---GRADE GENERATION vs QUESTION---
---DECISION: GENERATION ADDRESSES QUESTION---
"Node 'generate':"
'\n---\n'
('It is expected that the Chicago Bears could have the opportunity to draft '
 'the first defensive player in the 2024 NFL draft. The Bears have the first '
 'overall pick in the draft, giving them a prime position to select top '
 'talent. The top wide receiver Marvin Harrison Jr. from Ohio State is also '
 'mentioned as a potential pick for the Cardinals.')

跟踪

https://smith.langchain.com/public/7e3aa7e5-c51f-45c2-bc66-b34f17ff2263/r

在 [18]

# Run
inputs = {"question": "What are the types of agent memory?"}
for output in app.stream(inputs):
    for key, value in output.items():
        # Node
        pprint(f"Node '{key}':")
        # Optional: print full state at each node
        # pprint.pprint(value["keys"], indent=2, width=80, depth=None)
    pprint("\n---\n")

# Final generation
pprint(value["generation"])

# 运行 inputs = {"question": "代理内存的类型有哪些？"} for output in app.stream(inputs): for key, value in output.items(): # 节点 pprint(f"节点 '{key}':") # 可选：在每个节点打印完整状态 # pprint.pprint(value["keys"], indent=2, width=80, depth=None) pprint("\n---\n") # 最终生成 pprint(value["generation"])

---ROUTE QUESTION---
---ROUTE QUESTION TO RAG---
---RETRIEVE---
"Node 'retrieve':"
'\n---\n'
---CHECK DOCUMENT RELEVANCE TO QUESTION---
---GRADE: DOCUMENT RELEVANT---
---GRADE: DOCUMENT RELEVANT---
---GRADE: DOCUMENT NOT RELEVANT---
---GRADE: DOCUMENT RELEVANT---
---ASSESS GRADED DOCUMENTS---
---DECISION: GENERATE---
"Node 'grade_documents':"
'\n---\n'
---GENERATE---
---CHECK HALLUCINATIONS---
---DECISION: GENERATION IS GROUNDED IN DOCUMENTS---
---GRADE GENERATION vs QUESTION---
---DECISION: GENERATION ADDRESSES QUESTION---
"Node 'generate':"
'\n---\n'
('The types of agent memory include Sensory Memory, Short-Term Memory (STM) or '
 'Working Memory, and Long-Term Memory (LTM) with subtypes of Explicit / '
 'declarative memory and Implicit / procedural memory. Sensory memory retains '
 'sensory information briefly, STM stores information for cognitive tasks, and '
 'LTM stores information for a long time with different types of memories.')

跟踪

https://smith.langchain.com/public/fdf0a180-6d15-4d09-bb92-f84f2105ca51/r

评论