Langgraph从零开始构建第一个Agentic RAG 系统

智能体人工智能正在迅速普及，现在是时候进行另一次详细的教程了，这不仅能帮助你构建你的第一个智能体 RAG 系统，还能让你深入了解它的组成部分。

pip install sentence_transformerspip install langgraph==1.0.7pip install langchain_openai==1.1.7pip install langchain-core==1.2.7pip install google-search-results

我们将从头开始编写 RAG 和 Web 搜索工具。

# Set up our custom company vector databasecompany_db = VectorDatabase()# Hypothetical company informationcompany_information = [    "Quantum Horizons Inc. is a pioneering space exploration company founded in 2030.",    "The company specializes in developing quantum-powered spacecraft for interplanetary travel.",    "With a team of 500 aerospace engineers and quantum physicists, Quantum Horizons is pushing the boundaries of space technology.",    "Their flagship project, the 'StarLeap', aims to reduce travel time to Mars from months to just weeks.",    "Quantum Horizons has established the first permanent research base on the Moon's far side.",    "The company's innovative quantum propulsion system has revolutionized the concept of space travel.",    "Headquartered in a state-of-the-art facility in Houston, Quantum Horizons also maintains orbital research stations.",    "They've partnered with major space agencies worldwide to advance human presence in the solar system.",    "Quantum Horizons' CEO, Dr. Zara Novak, is a former astronaut and a leading expert in quantum mechanics.",    "The company's mission is to make interplanetary travel accessible and establish humanity as a multi-planet species."]

@tool("RAG Tool")def rag_tool(question: str) -> str:    """Tool to search for relevant information from a vector database."""    # Encode the question    query_vec = model.encode(question)    # Get top 5 similar vector    results = company_db.search(query_vec, top_k = 5)    # Build context from the results    context = "\n".join([f"- {res['metadata']['sentence']}" for res in results])    # Return the answer    return context

os.environ["SERPER_API_KEY"] = "YOUR_API_KEY"

# Create custom Web search tool@tooldef web_search_tool(query: str) -> str:    """    Use this tool to search the internet for CURRENT events, news, and external knowledge.    CRITICAL: You MUST use this tool for any questions regarding 2024, 2025, 2026,    Nobel prizes, sports results, or real-time information.    """    search = GoogleSearch({        # "as_dt":'exclude',#include        "q": query,        "api_key": os.environ.get("SERPER_API_KEY"),    })    results = search.get_dict()    # print(results)    # print(results['organic_results'])    # Get value associated with key "organic" else return []    search_results = results.get("organic_results", [])    if not search_results:        return "No search results found."    context = ""    for result in search_results:        title = result.get("title", "")        link = result.get("link", "")        snippet = result.get("snippet", "")        context += f"Title: {title}\nLink: {link}\nSnippet: {snippet}\n\n"    # print(context)    return f"Web Search Results:\n{context}"

llm = ChatOpenAI(    model=MODEL,    api_key=API_KEY,    base_url=API_URL,    temperature=0)tools = [rag_tool, web_search_tool]# 绑定工具到 LLM，这样 LLM 才知道可以调用它们llm_with_tools = llm.bind_tools(tools)# 临时测试代码# llm_with_tools = llm.bind_tools(tools)response = llm_with_tools.invoke([HumanMessage(content="Hello")])print(response)# ==========================================# 2. State (状态定义)# ==========================================class AgentState(TypedDict):    # 消息历史：存储所有对话记录（包括工具调用）    messages: Annotated[List[BaseMessage], add_messages]    # 用户原始查询    user_query: str    # 最终结果    final_answer: str# ==========================================# 3. Nodes (手动定义节点)# ==========================================# --- Node 1: Retriever Node (Core Logic) ---import datetime# ... (之前的代码)def retriever_node(state: AgentState):    messages = state["messages"]    current_date = datetime.datetime.now().strftime("%Y-%m-%d")    # --- 关键修改：将 Prompt 改为“回退策略 (Fallback Strategy)” ---    system_prompt_content = f"""You are a smart researcher assistant. Current Date: {current_date}.    Your Goal: Answer the user's question using the most reliable source.    ### EXECUTION STRATEGY (Follow this strictly):    1. **PHASE 1: Check Internal Knowledge (RAG)**       - For ANY information query (technical, company, factual), ALWAYS start by calling 'rag_tool' first.       - Exception: Only skip RAG if the user is just saying "Hi" or asking for help writing a generic email.    2. **PHASE 2: Evaluate & Fallback**       - Look at the output from 'rag_tool'.       - **IF** the RAG context contains the answer -> Finalize the answer.       - **IF** the RAG context is empty, irrelevant, or doesn't fully answer the question -> **YOU MUST IMMEDIATELY call 'web_search_tool'**.    3. **PHASE 3: Final Answer**       - Use the info from Web Search to answer.    ### CRITICAL RULES:    - Do NOT answer from your own memory if RAG fails. Search the web instead.    - If the user asks about "2024/2025/2026 events" or "News", and RAG fails, you MUST use web search.    User Query: {state['user_query']}    """    # 逻辑保持不变：如果没有 SystemMessage 则添加    if isinstance(messages[0], SystemMessage):        input_messages = [SystemMessage(content=system_prompt_content)] + messages[1:]    else:        input_messages = [SystemMessage(content=system_prompt_content)] + messages    # 调用 LLM    response = llm_with_tools.invoke(input_messages)    # --- Debug 信息 (方便你观察它是否进行了“二次尝试”) ---    print(f"\nDEBUG [Retriever]: Processing step...")    if response.tool_calls:        print(f"  -> Decision: Calling Tool: {[t['name'] for t in response.tool_calls]}")    else:        print(f"  -> Decision: Responding directly (Content length: {len(response.content)})")    return {"messages": [response]}# --- Node 2: Tool Execution Node ---# 使用 LangGraph 内置的 ToolNode 来运行工具，这是最稳定的组件tool_node = ToolNode(tools)# --- Node 3: Customer Support Node ---def customer_support_node(state: AgentState):    """    客服代理。    当 Retriever 决定停止搜索并给出总结后，这个节点接手进行最终润色。    """    # 获取 Retriever 的最后一条消息作为检索到的信息    last_message = state["messages"][-1]    retrieved_info = last_message.content    prompt = ChatPromptTemplate.from_messages([        ("system", (            "You are a Senior Customer Support Agent.\n"            "Goal: Answer the user query using the retrieved info accurately and concisely.\n"            "If info is missing, apologize."        )),        ("user", (            "User Query: {user_query}\n\n"            "Retrieved Context:\n{retrieved_info}"        ))    ])    chain = prompt | llm    response = chain.invoke({        "user_query": state["user_query"],        "retrieved_info": retrieved_info    })    return {"final_answer": response.content}# ==========================================# 4. Conditional Logic (路由逻辑)# ==========================================def route_retriever(state: AgentState):    """    决定下一步去哪：    1. 如果 LLM 想要调用工具 -> 去 'tools' 节点    2. 如果 LLM 输出了文本（找到了信息） -> 去 'customer_support' 节点    """    last_message = state["messages"][-1]    # 检查是否有工具调用请求    if last_message.tool_calls:        return "tools"    else:        return "customer_support"# ==========================================# 5. Graph Construction (构建图)# ==========================================workflow = StateGraph(AgentState)# 添加节点workflow.add_node("retriever", retriever_node)workflow.add_node("tools", tool_node)workflow.add_node("customer_support", customer_support_node)# 设置入口workflow.add_edge(START, "retriever")# 设置条件边 (从 Retriever 出发)workflow.add_conditional_edges(    "retriever",    route_retriever,    {        "tools": "tools",  # 如果决定调用工具，去 tools 节点        "customer_support": "customer_support"  # 如果决定结束搜索，去客服节点    })# 设置工具返回后的边 (工具执行完 -> 回到 Retriever 继续思考)workflow.add_edge("tools", "retriever")# 设置结束边workflow.add_edge("customer_support", END)# 编译app = workflow.compile()# ==========================================# 6. Run (运行)# ==========================================def run_query(query: str):    print(f"\n--- Running for: {query} ---")    # 初始化状态，放入用户的查询    inputs = {        "user_query": query,        "messages": [HumanMessage(content=query)]    }    try:        # 运行图        result = app.invoke(inputs)        print(f"Final Result:\n{result['final_answer']}")    except Exception as e:        import traceback        traceback.print_exc()if __name__ == "__main__":    # Test 1: 需要 RAG    run_query("What is the name of the flagship project of the company?")    # Test 2: 需要 Web Search    run_query("Who won the Nobel prize in 2024 in Physics?")

run_query("What is the name of the flagship project of the company?")

# Test 2: 需要 Web Searchrun_query("Who won the Nobel prize in 2024 in Physics?")