深度学习：构建基于LangGraph的RAG多智能体研究工具

（一）传统问答系统的局限性

（二）多智能体RAG的核心优势

小提示：这种功能使基于智能体的RAG系统成为企业应用中更稳健、更可靠的解决方案，因为在企业场景中，可靠性至关重要。

from docling.document_converter import DocumentConverter

logger.info("Starting document processing.")
converter = DocumentConverter()
markdown_document = converter.convert(source).document.export_to_markdown()

重要提示：通过OpenAI的嵌入模型，我们将文档列表转换为向量，并存储在Chroma中。

from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

embd = OpenAIEmbeddings()

vectorstore_from_documents = Chroma.from_documents(
    documents=docs_list,
    collection_name="rag-chroma-google-v1",
    embedding=embd,
    persist_directory='db_vector'
)

from pydantic import BaseModel, Field
from typing import Literal, TypedDict

class Router(TypedDict):
    """Classify user query."""
    logic: str
    type: Literal["more-info", "environmental", "general"]

class GradeHallucinations(BaseModel):
    """Binary score for hallucination present in generation answer."""
    binary_score: str = Field(description="Answer is grounded in the facts, '1' or '0'")

三、工作流程详解：八步搞定复杂查询

路由逻辑说明：工作流将被路由到例如"create_research_plan"、"ask_for_more_info"或"respond_to_general_query"等节点。

async def analyze_and_route_query(
    state: AgentState, *, config: RunnableConfig
) -> dict[str, Router]:
    """Analyze the user's query and determine the appropriate routing."""
    model = ChatOpenAI(model=GPT_4o, temperature=TEMPERATURE, streaming=True)
    messages = [
        {"role": "system", "content": ROUTER_SYSTEM_PROMPT}
    ] + state.messages
    logging.info("---ANALYZE AND ROUTE QUERY---")
    response = cast(
        Router, await model.with_structured_output(Router).ainvoke(messages)
    )
    return {"router": response}

async def create_research_plan(
    state: AgentState, *, config: RunnableConfig
) -> dict[str, list[str] | str]:
    """Create a step-by-step research plan for answering an environmental-related query."""
    class Plan(TypedDict):
        """Generate research plan."""
        steps: list[str]

    model = ChatOpenAI(model=GPT_4o_MINI, temperature=TEMPERATURE, streaming=True)
    messages = [
        {"role": "system", "content": RESEARCH_PLAN_SYSTEM_PROMPT}
    ] + state.messages
    logging.info("---PLAN GENERATION---")
    response = cast(Plan, await model.with_structured_output(Plan).ainvoke(messages))
    return {"steps": response["steps"], "documents": "delete"}

async def conduct_research(state: AgentState) -> dict[str, Any]:
    """Execute the first step of the research plan."""
    result = await researcher_graph.ainvoke({"question": state.steps[0]})  # graph call directly
    docs = result["documents"]
    step = state.steps[0]
    logging.info(f"n{len(docs)} documents retrieved in total for the step: {step}.")
    return {"documents": result["documents"], "steps": state.steps[1:]}

async def generate_queries(
    state: ResearcherState, *, config: RunnableConfig
) -> dict[str, list[str]]:
    """Generate search queries based on the question."""
    class Response(TypedDict):
        queries: list[str]

    logger.info("---GENERATE QUERIES---")
    model = ChatOpenAI(model="gpt-4o-mini-2024-07-18", temperature=0)
    messages = [
        {"role": "system", "content": GENERATE_QUERIES_SYSTEM_PROMPT},
        {"role": "human", "content": state.question},
    ]
    response = cast(Response, await model.with_structured_output(Response).ainvoke(messages))
    queries = response["queries"]
    queries.append(state.question)
    logger.info(f"Queries: {queries}")
    return {"queries": response["queries"]}

def _setup_vectorstore() -> Chroma:
    """Set up and return the Chroma vector store instance."""
    embeddings = OpenAIEmbeddings()
    return Chroma(
        collection_name=VECTORSTORE_COLLECTION,
        embedding_function=embeddings,
        persist_directory=VECTORSTORE_DIRECTORY
    )

# Create base retrievers
retriever_bm25 = BM25Retriever.from_documents(documents, search_kwargs={"k": TOP_K})
retriever_vanilla = vectorstore.as_retriever(search_type="similarity", search_kwargs={"k": TOP_K})
retriever_mmr = vectorstore.as_retriever(search_type="mmr", search_kwargs={"k": TOP_K})

ensemble_retriever = EnsembleRetriever(
    retrievers=[retriever_vanilla, retriever_mmr, retriever_bm25],
    weights=ENSEMBLE_WEIGHTS,
)

# Set up Cohere re-ranking
compressor = CohereRerank(top_n=2, model="rerank-english-v3.0")
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=ensemble_retriever,
)

技术亮点：这里使用了三种检索器的集成（相似度检索、MMR检索、BM25检索），再经过Cohere重排，大大提高了检索的准确性。

def check_finished(state: AgentState) -> Literal["respond", "conduct_research"]:
    """Determine if the research process is complete."""
    if len(state.steps or []) > 0:
        return "conduct_research"
    else:
        return "respond"

async def respond(
    state: AgentState, *, config: RunnableConfig
) -> dict[str, list[BaseMessage]]:
    """Generate the final response to the user's query."""
    model = ChatOpenAI(model="gpt-4o-2024-08-06", temperature=0)
    context = format_docs(state.documents)
    prompt = RESPONSE_SYSTEM_PROMPT.format(context=context)
    messages = [{"role": "system", "content": prompt}] + state.messages
    response = await model.ainvoke(messages)
    return {"messages": [response]}

async def check_hallucinations(
    state: AgentState, *, config: RunnableConfig
) -> dict[str, Any]:
    """Analyze the response for hallucinations."""
    model = ChatOpenAI(model=GPT_4o_MINI, temperature=TEMPERATURE, streaming=True)
    system_prompt = CHECK_HALLUCINATIONS.format(
        documents=state.documents,
        generation=state.messages[-1]
    )
    messages = [
        {"role": "system", "content": system_prompt}
    ] + state.messages
    logging.info("---CHECK HALLUCINATIONS---")
    response = cast(GradeHallucinations, await model.with_structured_output(GradeHallucinations).ainvoke(messages))
    return {"hallucination": response}

交互方式：用户可以选择仅重新执行最后的生成步骤，而无需重新启动整个工作流，或者选择结束流程。这种人工参与机制确保了用户对整个过程的控制，避免了不必要的循环或不期望的操作。

def human_approval(state: AgentState):
    _binary_score = state.hallucination.binary_score
    if _binary_score == "1":
        return"END"
    else:
        retry_generation = interrupt(
            {
                "question": "Is this correct?",
                "llm_output": state.messages[-1]
            }
        )
        if retry_generation == "y":
            print("Continue with retry...")
            return"respond"
        else:
            return"END"

四、实战测试：多智能体RAG的强大能力

未来展望：多智能体RAG是人工智能领域的一项重大突破。它将大型语言模型的能力与自主推理和信息检索相结合，引入了一种新的智能和灵活性标准。随着人工智能的不断发展，多智能体RAG将在各个行业中发挥基础性作用，彻底改变我们使用技术的方式。

你是一名 AI 行业编辑，请围绕下面这条热点输出一份资讯解读：
热点：深度学习：构建基于LangGraph的RAG多智能体研究工具要求：
1. 先用一句话解释这条热点在讲什么
2. 再总结它为什么重要
3. 说明会影响哪些 AI 产品或内容方向
4. 最后给出 3 个适合资讯站使用的标题