颠覆传统生成方式！Adaptive RAG：实时学习、智能调整的下一代检索增强技术

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 llm_fallback(state):    """    Generate answer using the LLM w/o vectorstore
    Args:        state (dict): The current graph state
    Returns:        state (dict): New key added to state, generation, that contains LLM generation    """    print("---LLM Fallback---")    question = state["question"]    generation = llm_chain.invoke({"question": question})    return {"question": question, "generation": generation}

def generate(state):    """    Generate answer using the vectorstore
    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"]    if not isinstance(documents, list):        documents = [documents]
    # RAG generation    generation = rag_chain.invoke({"documents": 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 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})
    # Fallback to LLM or raise error if no decision    if "tool_calls" not in source.additional_kwargs:        print("---ROUTE QUESTION TO LLM---")        return "llm_fallback"    if len(source.additional_kwargs["tool_calls"]) == 0:        raise "Router could not decide source"
    # Choose datasource    datasource = source.additional_kwargs["tool_calls"][0]["function"]["name"]    if datasource == "web_search":        print("---ROUTE QUESTION TO WEB SEARCH---")        return "web_search"    elif datasource == "vectorstore":        print("---ROUTE QUESTION TO RAG---")        return "vectorstore"    else:        print("---ROUTE QUESTION TO LLM---")        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, WEB SEARCH---")        return "web_search"    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"