LangGraph 实战：搭建带状态的多步骤 AI Agent

## 创建虚拟环境
python -m venv .venv
source .venv/bin/activate  # Windows 用 .venv\Scripts\activate

## 安装核心库
pip install langgraph langchain-openai duckduckgo-search

export OPENAI_API_KEY="sk-your-key-here"

from typing import TypedDict
from langgraph.graph import StateGraph, END

## 1. 定义状态
class AgentState(TypedDict):
    question: str
    answer: str

## 2. 定义节点：收状态、返回状态更新的函数
def answer_node(state: AgentState):
    return {"answer": f"你问的是：{state['question']}"}

## 3. 建图
graph = StateGraph(AgentState)
graph.add_node("answer", answer_node)
graph.set_entry_point("answer")   # 入口
graph.add_edge("answer", END)     # 固定走到终点
compiled = graph.compile()

## 4. 运行
result = compiled.invoke({"question": "Python 还是 Java？"})
print(result["answer"])

import os
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage
from duckduckgo_search import DDGS

## --- 状态定义 ---
class ResearchState(TypedDict):
    messages: list
    search_count: int
    final_answer: str

## --- 工具函数 ---
def web_search(query: str) -> str:
    """用 DuckDuckGo 搜索并返回前 3 条摘要"""
    results = DDGS().text(query, max_results=3)
    return "\n".join([f"- {r['title']}: {r['body']}" for r in results])

## --- 节点1: 决定下一步 ---
def decide_next(state: ResearchState) -> str:
    # 如果已经搜过 3 次，直接结束
    if state.get("search_count", 0) >= 3:
        return "summarize"
    # 否则继续搜索
    return "search"

## --- 节点2: 执行搜索 ---
def search_node(state: ResearchState) -> dict:
    last_msg = state["messages"][-1].content
    results = web_search(last_msg)
    return {
        "messages": [AIMessage(content=f"[搜索结果]\n{results}")],
        "search_count": state.get("search_count", 0) + 1
    }

## --- 节点3: 生成最终回答 ---
def summarize_node(state: ResearchState) -> dict:
    llm = ChatOpenAI(model="gpt-4o-mini")
    prompt = "基于以下搜索结果，生成一份结构化对比总结：\n" + state["messages"][-1].content
    response = llm.invoke([HumanMessage(content=prompt)])
    return {"final_answer": response.content}

## --- 建图 ---
graph = StateGraph(ResearchState)
graph.add_node("search", search_node)
graph.add_node("summarize", summarize_node)
graph.set_entry_point("search")

## 条件路由
graph.add_conditional_edges("search", decide_next, {
    "search": "search",
    "summarize": "summarize"
})
graph.add_edge("summarize", END)

workflow = graph.compile()

## --- 运行 ---
result = workflow.invoke({
    "messages": [HumanMessage(content="对比 Kafka 和 RabbitMQ 的核心差异")],
    "search_count": 0,
    "final_answer": ""
})
print(result["final_answer"])