高级RAG：纠正检索增强生成（CRAG）与LangGraph

! pip install langchain_community tiktoken langchain-openai langchainhub chromadb langchain langgraph tavily-python

import osos.environ['OPENAI_API_KEY'] = ""

os.environ['TAVILY_API_KEY'] = ""

os.environ['LANGCHAIN_TRACING_V2'] = 'true'os.environ['LANGCHAIN_ENDPOINT'] = 'https://api.smith.langchain.com' os.environ['LANGCHAIN_API_KEY'] = ""

from langchain.text_splitter import RecursiveCharacterTextSplitterfrom langchain_community.document_loaders import WebBaseLoaderfrom langchain_community.vectorstores import Chromafrom 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()

### 检索评分器
from langchain_openai import ChatOpenAIfrom langchain_core.prompts import ChatPromptTemplatefrom langchain_core.pydantic_v1 import BaseModel, Field
# 数据模型class GradeDocuments(BaseModel):"""对检索到的文档与用户问题的相关性进行二元评分检查的模型。"""
binary_score: str = Field(description="文档与问题相关，'是'或'否'")
# 带功能调用的LLMllm = ChatOpenAI(model="gpt-3.5-turbo-0125", temperature=0)structured_llm_grader = llm.with_structured_output(GradeDocuments)
# 提示system = """你是评估检索到的文档与用户问题相关性的评分员。\n 如果文档包含与问题相关的关键词或语义含义，将其评为相关。\n给出二元分数'是'或'否'，以指示文档是否与问题相关。"""grade_prompt = ChatPromptTemplate.from_messages([("system", system),("human", "检索到的文档：\n\n {document} \n\n 用户问题：{question}"),])
retrieval_grader = grade_prompt | structured_llm_graderquestion = "agent memory"docs = retriever.get_relevant_documents(question)doc_txt = docs[1].page_contentprint(retrieval_grader.invoke({"question": question, "document": doc_txt}))

### 生成
from langchain import hubfrom langchain_core.output_parsers import StrOutputParser
# 提示prompt = hub.pull("rlm/rag-prompt")
# LLMllm = 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)

### 问题重写器
# LLM llm = ChatOpenAI(model="gpt-3.5-turbo-0125", temperature=0)
# 提示 system = """你是一个将输入问题转换为优化的更好版本的问题的重写器，\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})

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

from typing_extensions import TypedDictfrom typing import List
class GraphState(TypedDict):"""表示我们的图的状态。
属性：question: 问题generation: LLM生成web_search: 是否添加搜索documents: 文档列表 """question : strgeneration : strweb_search : strdocuments : List[str]

from langchain.schema import Document
def retrieve(state):"""检索文档
参数：state (dict): 当前图状态
返回：state (dict): 向状态添加新键，文档，包含检索到的文档"""print("---检索---")question = state["question"]
# 检索documents = retriever.get_relevant_documents(question)return {"documents": documents, "question": question}
def generate(state):"""生成答案
参数：state (dict): 当前图状态
返回：state (dict): 向状态添加新键，生成，包含LLM生成"""print("---生成---")question = state["question"]documents = state["documents"]# RAG生成generation = rag_chain.invoke({"context": documents, "question": question})return {"documents": documents, "question": question, "generation": generation}
def grade_documents(state):"""确定检索到的文档是否与问题相关。
参数：state (dict): 当前图状态
返回：state (dict): 更新文档键，只包含过滤后的相关文档"""
print("---检查文档与问题的相关性---")question = state["question"]documents = state["documents"]# 为每个文档打分filtered_docs = []web_search = "否"for d in documents:score = retrieval_grader.invoke({"question": question, "document": d.page_content})grade = score.binary_scoreif grade == "是":print("---评分：文档相关---")filtered_docs.append(d)else:print("---评分：文档不相关---")web_search = "是"continuereturn {"documents": filtered_docs, "question": question, "web_search": web_search}
def transform_query(state):"""转换查询以产生更好的问题。
参数：state (dict): 当前图状态
返回：state (dict): 更新问题键，用重述的问题更新"""
print("---转换查询---")question = state["question"]documents = state["documents"]
# 重写问题better_question = question_rewriter.invoke({"question": question})return {"documents": documents, "question": better_question}def web_search(state):"""基于重述的问题进行网页搜索。
参数：state (dict): 当前图状态
返回：state (dict): 更新文档键，附加网页结果"""
print("---网页搜索---")question = state["question"]documents = state["documents"]
# 网页搜索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.append(web_results)
return {"documents": documents, "question": question}
### 边
def decide_to_generate(state):"""确定是否生成答案，或者重新生成问题。
参数：state (dict): 当前图状态
返回：str: 下一个节点调用的二元决策"""
print("---评估评分文档---")question = state["question"]web_search = state["web_search"]filtered_documents = state["documents"]
if web_search == "是":# 所有文档都已通过check_relevance过滤# 我们将重新生成新查询print("---决策：所有文档与问题不相关，转换查询---")return "transform_query"else:# 我们有相关文档，所以生成答案print("---决策：生成---")return "generate"

from langgraph.graph import END, StateGraph
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.set_entry_point("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"节点 '{key}':")# 可选：在每个节点打印完整状态# pprint.pprint(value["keys"], indent=2, width=80, depth=None)pprint("\n---\n")
# 最终生成pprint(value["generation"])

# — -RETRIEVE — -“Node ‘retrieve’:”‘\n — -\n’— -CHECK DOCUMENT RELEVANCE TO QUESTION — -— -GRADE: DOCUMENT NOT RELEVANT — -— -GRADE: DOCUMENT NOT RELEVANT — -— -GRADE: DOCUMENT RELEVANT — -— -GRADE: DOCUMENT RELEVANT — -“Node ‘grade_documents’:”‘\n — -\n’— -ASSESS GRADED DOCUMENTS — -— -DECISION: ALL DOCUMENTS ARE NOT RELEVANT TO QUESTION, TRANSFORM QUERY — -— -TRANSFORM QUERY — -“Node ‘transform_query’:”‘\n — -\n’— -WEB SEARCH — -“Node ‘web_search_node’:”‘\n — -\n’— -GENERATE — -“Node ‘generate’:”‘\n — -\n’“Node ‘__end__’:”‘\n — -\n’(‘Agents possess short-term memory, which is utilized for in-context learning, ‘‘and long-term memory, allowing them to retain and recall vast amounts of ‘‘information over extended periods. Some experts also classify working memory ‘‘as a distinct type, although it can be considered a part of short-term ‘‘memory in many cases.’)

from pprint import pprint
# 运行inputs = {"question": "How does the AlphaCodium paper work?"}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"])

# — -RETRIEVE — -“Node ‘retrieve’:”‘\n — -\n’— -CHECK DOCUMENT RELEVANCE TO QUESTION — -— -GRADE: DOCUMENT NOT RELEVANT — -— -GRADE: DOCUMENT NOT RELEVANT — -— -GRADE: DOCUMENT NOT RELEVANT — -— -GRADE: DOCUMENT RELEVANT — -“Node ‘grade_documents’:”‘\n — -\n’— -ASSESS GRADED DOCUMENTS — -— -DECISION: ALL DOCUMENTS ARE NOT RELEVANT TO QUESTION, TRANSFORM QUERY — -— -TRANSFORM QUERY — -“Node ‘transform_query’:”‘\n — -\n’— -WEB SEARCH — -“Node ‘web_search_node’:”‘\n — -\n’— -GENERATE — -“Node ‘generate’:”‘\n — -\n’“Node ‘__end__’:”‘\n — -\n’(‘The AlphaCodium paper functions by proposing a code-oriented iterative flow ‘‘that involves repeatedly running and fixing generated code against ‘‘input-output tests. Its key mechanisms include generating additional data ‘‘like problem reflection and test reasoning to aid the iterative process, as ‘‘well as enriching the code generation process. AlphaCodium aims to improve ‘‘the performance of Large Language Models on code problems by following a ‘‘test-based, multi-stage approach.’)