📊 LangGraph 与 Agent 开发

LangChain Agent 的局限

LangChain 提供了早期的 Agent 实现（如 AgentExecutor），但它有一些根本性的局限：

• 线性流程：AgentExecutor 本质是一个循环——思考→行动→观察→思考，难以表达复杂的分支逻辑
• 黑盒执行：Agent 的决策路径不透明，调试困难，出问题时难以定位
• 状态管理缺失：无法在多个节点间共享和持久化复杂状态
• 人机交互受限：缺乏在流程中途暂停等待人工审批的机制
• 难以组合：无法将多个 Agent 组合成更复杂的工作流

核心优势

1️⃣ State（状态）

State 是 LangGraph 中所有节点共享的数据结构，使用 Python 的 TypedDict 或 Pydantic 的 BaseModel 定义。

Pythonfrom typing import TypedDict, Annotated, List
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    """Agent 的共享状态"""
    messages: Annotated[List, add_messages]  # 消息列表，自动累积
    query: str                             # 用户查询
    documents: List[str]                   # 检索到的文档
    tool_calls: int                        # 工具调用次数

2️⃣ Node（节点）

Node 是一个普通 Python 函数，接收当前 State，返回状态更新（Partial State）：

Pythondef llm_node(state: AgentState) -> dict:
    """LLM 节点：调用大模型生成回复"""
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

def tool_node(state: AgentState) -> dict:
    """工具节点：执行工具调用"""
    last_message = state["messages"][-1]
    results = [execute_tool(tc) for tc in last_message.tool_calls]
    return {"messages": results}

3️⃣ Edge（边）

Edge 定义节点之间的固定连接关系，表示无条件的流转：

Python# 固定边：从 start 直接进入 llm_node
graph.add_edge("start", "llm_node")

# 固定边：从 tool_node 回到 llm_node
graph.add_edge("tool_node", "llm_node")

5️⃣ StateGraph（图构建器）

StateGraph 是图的顶层构建器，负责组合所有节点和边：

Pythonfrom langgraph.graph import StateGraph, START, END

# 创建图
graph = StateGraph(AgentState)

# 添加节点
graph.add_node("llm_node", llm_node)
graph.add_node("tool_node", tool_node)

# 添加边
graph.add_edge(START, "llm_node")
graph.add_conditional_edges("llm_node", should_continue, ...)
graph.add_edge("tool_node", "llm_node")

# 编译为可执行的图
app = graph.compile()

运行方式

编译后的图支持三种运行模式：

Python — 运行示例# 方式 1: 同步调用（等待完整结果）
result = app.invoke({"messages": ["北京今天天气如何？"]})
print(result["messages"][-1].content)

# 方式 2: 流式输出（逐步获取每个节点的输出）
for chunk in app.stream({"messages": ["北京今天天气如何？"]}):
    print(chunk)  # 每个节点的输出

# 方式 3: 异步流式（适合 Web 服务）
async for chunk in app.astream({"messages": ["北京今天天气如何？"]}):
    print(chunk)

完整代码

Python — react_agent.pyfrom typing import TypedDict, Annotated, Literal
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode

# ═══ 1. 定义 State ═══
class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

# ═══ 2. 定义工具 ═══
@tool
def search(query: str) -> str:
    """搜索互联网获取最新信息"""
    # 实际项目中接入搜索引擎 API
    return f"搜索结果：关于 '{query}' 的最新信息..."

@tool
def calculator(expression: str) -> str:
    """计算数学表达式"""
    try:
        result = eval(expression)
        return str(result)
    except Exception as e:
        return f"计算错误: {e}"

tools = [search, calculator]
llm = ChatOpenAI(model="gpt-4o").bind_tools(tools)

# ═══ 3. 创建 LLM 节点 ═══
def llm_node(state: AgentState) -> dict:
    """调用 LLM 生成回复或决定使用工具"""
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

# ═══ 4. 创建工具节点 ═══
tool_node = ToolNode(tools)

# ═══ 5. 创建条件边 ═══
def should_continue(
    state: AgentState,
) -> Literal["tools", "end"]:
    """判断是否继续调用工具"""
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"
    return "end"

# ═══ 6. 构建图 ═══
graph = StateGraph(AgentState)
graph.add_node("agent", llm_node)
graph.add_node("tools", tool_node)

graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue, {
    "tools": "tools",
    "end": END,
})
graph.add_edge("tools", "agent")  # 工具执行完回到 LLM

# ═══ 7. 编译并运行 ═══
app = graph.compile()

# 运行 Agent
result = app.invoke({
    "messages": [HumanMessage(content="搜索 Python 最新版本，然后计算 123 * 456")]
})

print(result["messages"][-1].content)

代码解读

完整代码

Python — rag_agent.pyfrom typing import TypedDict, Annotated, Literal, List
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

# ═══ 1. 定义 State ═══
class RAGState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    query: str
    rewritten_query: str
    documents: List[str]

# ═══ 2. 初始化组件 ═══
llm = ChatOpenAI(model="gpt-4o")
embeddings = OpenAIEmbeddings()
vectorstore = FAISS.load_local("./faiss_index", embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

# ═══ 3. 路由节点：判断是否需要检索 ═══
def route_query(state: RAGState) -> Literal["rewrite_query", "generate"]:
    """判断问题是需要检索知识库，还是可以直接回答"""
    query = state["messages"][-1].content
    response = llm.invoke([
        "系统": "判断以下问题是否需要检索知识库。只需回答 YES 或 NO。",
        "用户": query,
    ])
    if "YES" in response.content:
        return "rewrite_query"
    return "generate"

# ═══ 4. 重写查询节点 ═══
def rewrite_query(state: RAGState) -> dict:
    """优化查询以提高检索质量"""
    query = state["messages"][-1].content
    response = llm.invoke([
        "系统": "将以下问题改写为更适合向量检索的关键词查询。只输出改写后的查询。",
        "用户": query,
    ])
    return {"rewritten_query": response.content}

# ═══ 5. 检索节点 ═══
def retrieve(state: RAGState) -> dict:
    """从向量数据库检索相关文档"""
    query = state.get("rewritten_query", state["messages"][-1].content)
    docs = retriever.invoke(query)
    return {"documents": [doc.page_content for doc in docs]}

# ═══ 6. 生成节点 ═══
def generate(state: RAGState) -> dict:
    """基于检索结果（如有）生成最终回答"""
    query = state["messages"][-1].content
    docs = state.get("documents", [])

    context = "\n\n".join(docs) if docs else "无相关文档。"
    response = llm.invoke([
        "系统": f"基于以下上下文回答用户问题。\n\n上下文：\n{context}",
        "用户": query,
    ])
    return {"messages": [response]}

# ═══ 7. 构建图 ═══
graph = StateGraph(RAGState)
graph.add_node("rewrite_query", rewrite_query)
graph.add_node("retrieve", retrieve)
graph.add_node("generate", generate)

graph.add_conditional_edges(START, route_query, {
    "rewrite_query": "rewrite_query",
    "generate": "generate",
})
graph.add_edge("rewrite_query", "retrieve")
graph.add_edge("retrieve", "generate")
graph.add_edge("generate", END)

# ═══ 8. 编译并运行 ═══
app = graph.compile()
result = app.invoke({
    "messages": [HumanMessage(content="什么是 RAG？")],
    "query": "",
    "rewritten_query": "",
    "documents": [],
})
print(result["messages"][-1].content)

完整代码

Python — customer_service_agent.pyfrom typing import TypedDict, Annotated, Literal, List
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

# ═══ State ═══
class CustomerServiceState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    intent: str               # 用户意图
    satisfaction: bool          # 用户满意度
    ticket_id: str             # 工单号
    conversation_history: List[dict]  # 多轮对话记录

llm = ChatOpenAI(model="gpt-4o")

# ═══ 1. 意图识别 ═══
def classify_intent(state: CustomerServiceState) -> dict:
    """识别用户意图：FAQ / 知识检索 / 工单创建 / 转人工"""
    last_msg = state["messages"][-1].content
    response = llm.invoke([
        {"role": "system", "content": """
        将用户消息分类为以下意图之一，只输出意图标签：
        - faq: 常见问题（退换货、物流、价格等常见咨询）
        - knowledge: 需要检索知识库的专业问题
        - ticket: 需要创建工单的问题（投诉、技术故障等）
        - human: 明确要求转人工或无法自动处理
        """},
        {"role": "user", "content": last_msg},
    ])
    intent = response.content.strip().lower()
    return {"intent": intent}

# ═══ 2. FAQ 回答 ═══
def faq_answer(state: CustomerServiceState) -> dict:
    """基于 FAQ 知识库回答常见问题"""
    faqs = {
        "退货": "您可以在订单完成后 7 天内申请退货，请前往「我的订单」→「申请退货」。",
        "物流": "您可以在「我的订单」中查看物流状态，一般 3-5 个工作日送达。",
        "价格": "我们的价格会根据活动进行调整，请关注首页的优惠信息。",
    }
    query = state["messages"][-1].content
    # 简单匹配，实际项目中用向量检索
    for keyword, answer in faqs.items():
        if keyword in query:
            return {"messages": [AIMessage(content=answer)]}
    return {"messages": [AIMessage(content="抱歉，我暂时无法回答这个问题，为您转接人工客服。")]}

# ═══ 3. 知识库检索 ═══
def knowledge_search(state: CustomerServiceState) -> dict:
    """从知识库检索并生成回答"""
    query = state["messages"][-1].content
    # 实际项目：使用 retriever 检索
    docs = retriever.invoke(query)
    context = "\n".join([doc.page_content for doc in docs])
    response = llm.invoke([
        {"role": "system", "content": f"基于以下知识库内容回答用户问题：\n{context}"},
        {"role": "user", "content": query},
    ])
    return {"messages": [response]}

# ═══ 4. 创建工单 ═══
def create_ticket(state: CustomerServiceState) -> dict:
    """创建客服工单"""
    query = state["messages"][-1].content
    # 实际项目：调用工单系统 API
    ticket_id = f"TK-{datetime.now().strftime('%Y%m%d%H%M%S')}"
    response = llm.invoke([
        {"role": "system", "content": "你是一个客服助手。告知用户工单已创建，并提供工单号。"},
        {"role": "user", "content": query},
    ])
    return {
        "messages": [response],
        "ticket_id": ticket_id,
    }

# ═══ 5. 满意度检查 ═══
def check_satisfaction(state: CustomerServiceState) -> Literal["end", "escalate"]:
    """检查用户是否满意回答"""
    last_msg = state["messages"][-1].content
    response = llm.invoke([
        {"role": "system", "content": "判断用户对上一个回答是否满意。只回答 satisfied 或 unsatisfied。"},
        {"role": "user", "content": last_msg},
    ])
    if "satisfied" in response.content.lower():
        return "end"
    return "escalate"

# ═══ 6. 路由函数 ═══
def route_intent(state: CustomerServiceState) -> str:
    intent = state["intent"]
    routing = {
        "faq": "faq_answer",
        "knowledge": "knowledge_search",
        "ticket": "create_ticket",
        "human": "escalate",
    }
    return routing.get(intent, "faq_answer")

# ═══ 7. 构建图 ═══
graph = StateGraph(CustomerServiceState)
graph.add_node("classify_intent", classify_intent)
graph.add_node("faq_answer", faq_answer)
graph.add_node("knowledge_search", knowledge_search)
graph.add_node("create_ticket", create_ticket)
graph.add_node("check_satisfaction", check_satisfaction)

graph.add_edge(START, "classify_intent")
graph.add_conditional_edges("classify_intent", route_intent, {
    "faq_answer": "faq_answer",
    "knowledge_search": "knowledge_search",
    "create_ticket": "create_ticket",
    "escalate": "check_satisfaction",
})
graph.add_edge("faq_answer", "check_satisfaction")
graph.add_edge("knowledge_search", "check_satisfaction")
graph.add_edge("create_ticket", "check_satisfaction")
graph.add_conditional_edges("check_satisfaction", check_satisfaction, {
    "end": END,
    "escalate": "escalate",  # 可接入人工客服节点
})

# ═══ 8. 编译运行 ═══
app = graph.compile(checkpointer=MemorySaver())

# 多轮对话
config = {"configurable": {"thread_id": "customer-001"}}
print(app.invoke({
    "messages": [HumanMessage(content="我想退货")],
    "intent": "", "satisfaction": False, "ticket_id": "",
    "conversation_history": [],
}, config))

项目扩展建议