--- description: RAG --- # 使用本地语言模型机器（LLMs）的校正RAG（CRAG）校正RAG（CRAG）是RAG的一种策略，它在检索到的文档上结合了自反射/自分级。论文遵循以下一般流程： * 如果至少有一个文档超过相关性阈值，则继续生成 * 如果所有文档都低于相关性阈值，或者评分者不确定，则使用网页搜索来补充检索 * 在生成之前，对搜索或检索的文档进行知识精炼 * 将文档分割为知识条 * 对每个知识条进行评分，并过滤掉不相关的条目我们将使用 LangGraph 从头开始实现一些这些想法： * 如果有任何文档不相关，我们将使用网页搜索来补充检索。 * 我们将跳过知识精炼阶段，但如果需要的话，可以将其作为一个节点添加回去。 * 我们将使用 Tavily Search 进行网页搜索。

环境我们将使用以下工具和资源： * 使用 Ollama 应用程序访问本地语言模型（LLM）： * 下载 Ollama 应用程序。 * 拉取您选择的模型，例如：`ollama pull llama3` * 使用 Tavily 进行网页搜索。 * 使用 Nomic 的本地嵌入式向量存储，或者选择性地使用 OpenAI 的嵌入式向量存储。 * 使用 LangSmith 进行追踪和评估。 ``` %%capture --no-stderr %pip install -U langchain_community tiktoken langchainhub scikit-learn langchain langgraph tavily-python nomic[local] langchain-nomic langchain_openai ``` ```python # Search import os os.environ["TAVILY_API_KEY"] = "xxx" ``` ```python # Embedding (optional) os.environ["OPENAI_API_KEY"] = "xxx" ``` ```python # Tracing and testing (optional) os.environ["LANGCHAIN_API_KEY"] = "xxx" os.environ["LANGCHAIN_TRACING_V2"] = "true" os.environ["LANGCHAIN_ENDPOINT"] = "https://api.smith.langchain.com" os.environ["LANGCHAIN_PROJECT"] = "corrective-rag-agent-testing" ``` #### LLM 您可以从 Ollama 的语言模型（LLMs）中进行选择。 ```python local_llm = "llama3" model_tested = "llama3-8b" metadata = f"CRAG, {model_tested}" ``` #### 索引让我们索引3篇博文。 ```python from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_community.document_loaders import WebBaseLoader from langchain_community.vectorstores import SKLearnVectorStore from langchain_nomic.embeddings import NomicEmbeddings # local from langchain_openai import OpenAIEmbeddings # api # List of URLs to load documents from 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 documents from the URLs docs = [WebBaseLoader(url).load() for url in urls] docs_list = [item for sublist in docs for item in sublist] # Initialize a text splitter with specified chunk size and overlap text_splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder( chunk_size=250, chunk_overlap=0 ) # Split the documents into chunks doc_splits = text_splitter.split_documents(docs_list) # Embedding ''' embedding=NomicEmbeddings( model="nomic-embed-text-v1.5", inference_mode="local", ) ''' embedding = OpenAIEmbeddings() # Add the document chunks to the "vector store" vectorstore = SKLearnVectorStore.from_documents( documents=doc_splits, embedding=embedding, ) retriever = vectorstore.as_retriever(k=4) ``` #### Tools 工具 ```python ### Retrieval Grader from langchain.prompts import PromptTemplate from langchain_community.chat_models import ChatOllama from langchain_core.output_parsers import JsonOutputParser from langchain_mistralai.chat_models import ChatMistralAI # LLM llm = ChatOllama(model=local_llm, format="json", temperature=0) # Prompt prompt = PromptTemplate( template="""You are a teacher grading a quiz. You will be given: 1/ a QUESTION 2/ A FACT provided by the student You are grading RELEVANCE RECALL: A score of 1 means that ANY of the statements in the FACT are relevant to the QUESTION. A score of 0 means that NONE of the statements in the FACT are relevant to the QUESTION. 1 is the highest (best) score. 0 is the lowest score you can give. Explain your reasoning in a step-by-step manner. Ensure your reasoning and conclusion are correct. Avoid simply stating the correct answer at the outset. Question: {question} \n Fact: \n\n {documents} \n\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", "documents"], ) retrieval_grader = prompt | llm | JsonOutputParser() question = "agent memory" docs = retriever.invoke(question) doc_txt = docs[1].page_content print(retrieval_grader.invoke({"question": question, "documents": doc_txt})) ``` ``` {'score': '1'} ``` ```python ### Generate from langchain_core.output_parsers import StrOutputParser # Prompt prompt = PromptTemplate( template="""You are an assistant for question-answering tasks. Use the following documents to answer the question. If you don't know the answer, just say that you don't know. Use three sentences maximum and keep the answer concise: Question: {question} Documents: {documents} Answer: """, input_variables=["question", "documents"], ) # LLM llm = ChatOllama(model=local_llm, temperature=0) # Chain rag_chain = prompt | llm | StrOutputParser() # Run generation = rag_chain.invoke({"documents": docs, "question": question}) print(generation) ``` ``` The document mentions "memory stream" which is a long-term memory module that records a comprehensive list of agents' experience in natural language. It also discusses short-term memory and long-term memory, with the latter providing the agent with the capability to retain and recall information over extended periods. Additionally, it mentions planning and reflection mechanisms that enable agents to behave conditioned on past experience. ``` ```python ### Search from langchain_community.tools.tavily_search import TavilySearchResults web_search_tool = TavilySearchResults(k=3) ``` 图在这里，我们将明确定义大部分控制流程，仅在评分后使用LLM来定义一个单一的分支点。 ```python from typing import List from typing_extensions import TypedDict from IPython.display import Image, display from langchain.schema import Document from langgraph.graph import START, END, StateGraph class GraphState(TypedDict): """ Represents the state of our graph. Attributes: question: question generation: LLM generation search: whether to add search documents: list of documents """ question: str generation: str search: str documents: List[str] steps: List[str] 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 """ question = state["question"] documents = retriever.invoke(question) steps = state["steps"] steps.append("retrieve_documents") return {"documents": documents, "question": question, "steps": steps} 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 """ question = state["question"] documents = state["documents"] generation = rag_chain.invoke({"documents": documents, "question": question}) steps = state["steps"] steps.append("generate_answer") return { "documents": documents, "question": question, "generation": generation, "steps": steps, } 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 """ question = state["question"] documents = state["documents"] steps = state["steps"] steps.append("grade_document_retrieval") filtered_docs = [] search = "No" for d in documents: score = retrieval_grader.invoke( {"question": question, "documents": d.page_content} ) grade = score["score"] if grade == "yes": filtered_docs.append(d) else: search = "Yes" continue return { "documents": filtered_docs, "question": question, "search": search, "steps": steps, } 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 """ question = state["question"] documents = state.get("documents", []) steps = state["steps"] steps.append("web_search") web_results = web_search_tool.invoke({"query": question}) documents.extend( [ Document(page_content=d["content"], metadata={"url": d["url"]}) for d in web_results ] ) return {"documents": documents, "question": question, "steps": steps} 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 """ search = state["search"] if search == "Yes": return "search" else: return "generate" # Graph workflow = StateGraph(GraphState) # Define the nodes workflow.add_node("retrieve", retrieve) # retrieve workflow.add_node("grade_documents", grade_documents) # grade documents workflow.add_node("generate", generate) # generatae workflow.add_node("web_search", web_search) # web search # Build graph workflow.set_entry_point("retrieve") workflow.add_edge("retrieve", "grade_documents") workflow.add_conditional_edges( "grade_documents", decide_to_generate, { "search": "web_search", "generate": "generate", }, ) workflow.add_edge("web_search", "generate") workflow.add_edge("generate", END) custom_graph = workflow.compile() display(Image(custom_graph.get_graph(xray=True).draw_mermaid_png())) ```

```python import uuid def predict_custom_agent_local_answer(example: dict): config = {"configurable": {"thread_id": str(uuid.uuid4())}} state_dict = custom_graph.invoke( {"question": example["input"], "steps": []}, config ) return {"response": state_dict["generation"], "steps": state_dict["steps"]} example = {"input": "What are the types of agent memory?"} response = predict_custom_agent_local_answer(example) response ``` ``` {'response': 'According to the documents, there are two types of agent memory:\n\n* Short-term memory (STM): This is a data structure that holds information temporarily and allows the agent to process it when needed.\n* Long-term memory (LTM): This provides the agent with the capability to retain and recall information over extended periods.\n\nThese types of memories allow the agent to learn, reason, and make decisions.', 'steps': ['retrieve_documents', 'grade_document_retrieval', 'web_search', 'generate_answer']} ``` ### 评估现在我们已经定义了两种大致相同的代理架构！我们可以对它们进行评估。查看我们的概念指南以获取有关代理评估背景的内容。首先，我们可以评估代理在一组问题-答案对上的表现。我们将创建一个数据集并将其保存在 LangSmith 中。 ```python from langsmith import Client client = Client() # Create a dataset examples = [ ( "How does the ReAct agent use self-reflection? ", "ReAct integrates reasoning and acting, performing actions - such tools like Wikipedia search API - and then observing / reasoning about the tool outputs.", ), ( "What are the types of biases that can arise with few-shot prompting?", "The biases that can arise with few-shot prompting include (1) Majority label bias, (2) Recency bias, and (3) Common token bias.", ), ( "What are five types of adversarial attacks?", "Five types of adversarial attacks are (1) Token manipulation, (2) Gradient based attack, (3) Jailbreak prompting, (4) Human red-teaming, (5) Model red-teaming.", ), ( "Who did the Chicago Bears draft first in the 2024 NFL draft”?", "The Chicago Bears drafted Caleb Williams first in the 2024 NFL draft.", ), ("Who won the 2024 NBA finals?", "The Boston Celtics on the 2024 NBA finals"), ] # Save it dataset_name = "Corrective RAG Agent Testing" if not client.has_dataset(dataset_name=dataset_name): dataset = client.create_dataset(dataset_name=dataset_name) inputs, outputs = zip( *[({"input": text}, {"output": label}) for text, label in examples] ) client.create_examples(inputs=inputs, outputs=outputs, dataset_id=dataset.id) ``` 现在，我们将使用LLM作为分级器，将两种代理响应与地面实况参考进行比较answer.Here是我们可以使用的默认提示。我们将使用gpt-4o作为LLM分级器。 ```python from langchain import hub from langchain_openai import ChatOpenAI # Grade prompt grade_prompt_answer_accuracy = hub.pull("langchain-ai/rag-answer-vs-reference") def answer_evaluator(run, example) -> dict: """ A simple evaluator for RAG answer accuracy """ # Get the question, the ground truth reference answer, RAG chain answer prediction input_question = example.inputs["input"] reference = example.outputs["output"] prediction = run.outputs["response"] # Define an LLM grader llm = ChatOpenAI(model="gpt-4o", temperature=0) answer_grader = grade_prompt_answer_accuracy | llm # Run evaluator score = answer_grader.invoke( { "question": input_question, "correct_answer": reference, "student_answer": prediction, } ) score = score["Score"] return {"key": "answer_v_reference_score", "score": score} ``` ### 轨迹其次，我们可以评估每个代理相对于预期轨迹进行的工具调用列表。这会评估我们的代理所采取的特定推理轨迹！ ```python from langsmith.schemas import Example, Run # Reasoning traces that we expect the agents to take expected_trajectory_1 = [ "retrieve_documents", "grade_document_retrieval", "web_search", "generate_answer", ] expected_trajectory_2 = [ "retrieve_documents", "grade_document_retrieval", "generate_answer", ] def check_trajectory_react(root_run: Run, example: Example) -> dict: """ Check if all expected tools are called in exact order and without any additional tool calls. """ messages = root_run.outputs["messages"] tool_calls = find_tool_calls_react(messages) print(f"Tool calls ReAct agent: {tool_calls}") if tool_calls == expected_trajectory_1 or tool_calls == expected_trajectory_2: score = 1 else: score = 0 return {"score": int(score), "key": "tool_calls_in_exact_order"} def check_trajectory_custom(root_run: Run, example: Example) -> dict: """ Check if all expected tools are called in exact order and without any additional tool calls. """ tool_calls = root_run.outputs["steps"] print(f"Tool calls custom agent: {tool_calls}") if tool_calls == expected_trajectory_1 or tool_calls == expected_trajectory_2: score = 1 else: score = 0 return {"score": int(score), "key": "tool_calls_in_exact_order"} ``` ```python from langsmith.evaluation import evaluate experiment_prefix = f"custom-agent-{model_tested}" experiment_results = evaluate( predict_custom_agent_local_answer, data=dataset_name, evaluators=[answer_evaluator, check_trajectory_custom], experiment_prefix=experiment_prefix + "-answer-and-tool-use", num_repetitions=3, max_concurrency=1, # Use when running locally metadata={"version": metadata}, ) ``` ``` View the evaluation results for experiment: 'custom-agent-llama3-8b-answer-and-tool-use-d6006159' at: https://smith.langchain.com/o/1fa8b1f4-fcb9-4072-9aa9-983e35ad61b8/datasets/a8b9273b-ca33-4e2f-9f69-9bbc37f6f51b/compare?selectedSessions=83c60822-ef22-43e8-ac85-4488af279c6f ``` ``` 0it [00:00, ?it/s] ``` ``` Tool calls custom agent: ['retrieve_documents', 'grade_document_retrieval', 'web_search', 'generate_answer'] Tool calls custom agent: ['retrieve_documents', 'grade_document_retrieval', 'web_search', 'generate_answer'] Tool calls custom agent: ['retrieve_documents', 'grade_document_retrieval', 'web_search', 'generate_answer'] Tool calls custom agent: ['retrieve_documents', 'grade_document_retrieval', 'web_search', 'generate_answer'] Tool calls custom agent: ['retrieve_documents', 'grade_document_retrieval', 'web_search', 'generate_answer'] Tool calls custom agent: ['retrieve_documents', 'grade_document_retrieval', 'web_search', 'generate_answer'] Tool calls custom agent: ['retrieve_documents', 'grade_document_retrieval', 'web_search', 'generate_answer'] Tool calls custom agent: ['retrieve_documents', 'grade_document_retrieval', 'web_search', 'generate_answer'] Tool calls custom agent: ['retrieve_documents', 'grade_document_retrieval', 'web_search', 'generate_answer'] Tool calls custom agent: ['retrieve_documents', 'grade_document_retrieval', 'web_search', 'generate_answer'] Tool calls custom agent: ['retrieve_documents', 'grade_document_retrieval', 'web_search', 'generate_answer'] Tool calls custom agent: ['retrieve_documents', 'grade_document_retrieval', 'web_search', 'generate_answer'] Tool calls custom agent: ['retrieve_documents', 'grade_document_retrieval', 'web_search', 'generate_answer'] Tool calls custom agent: ['retrieve_documents', 'grade_document_retrieval', 'web_search', 'generate_answer'] Tool calls custom agent: ['retrieve_documents', 'grade_document_retrieval', 'web_search', 'generate_answer'] ``` 我们可以看到使用自定义代理（如图所示）和ReAct对GPT-4o和Llama-3-70b进行基准测试的结果。

本地自定义代理在工具调用可靠性方面表现良好：它遵循预期的推理轨迹。然而，答案准确性性能落后于具有自定义代理实现的较大模型。

LangGraph 使用本地语言模型机器（LLMs）的校正RAG（CRAG）