📌 目标
1️⃣ 案例一：智能客服工单系统
系统架构
┌─────────────────────────────────────────────────────────────┐│ 智能客服工单系统 │├─────────────────────────────────────────────────────────────┤│ 用户问题 ││ ↓ ││ [问题分类] ││ / | \ ││ 技术问题 退款 一般问题 ││ \ | / ││ END │└─────────────────────────────────────────────────────────────┘
核心代码
💡 条件路由基础已在 03_条件分支与循环 中详细讲解
from langgraph.graph import StateGraph, START, END, MessagesStatefrom langchain_core.messages import HumanMessage, AIMessagefrom typing import Annotated, Literalimport operatorclass CustomerServiceState(MessagesState): ticket_id: str category: str priority: Literal["low", "medium", "high"]def classify_issue(state: CustomerServiceState): """分类客户问题""" last_msg = state["messages"][-1].content.lower() if any(word in last_msg for word in ["紧急", "无法使用", "崩溃"]): category, priority = "technical", "high" elif any(word in last_msg for word in ["退款", "退货"]): category, priority = "refund", "medium" else: category, priority = "general", "low" return {"category": category, "priority": priority}def handle_technical(state: CustomerServiceState): return {"messages": [AIMessage(content="🔧 技术支持已接手...")]}def handle_refund(state: CustomerServiceState): return {"messages": [AIMessage(content="💰 退款部门已接收...")]}def handle_general(state: CustomerServiceState): return {"messages": [AIMessage(content="ℹ️ 客服团队已接收...")]}def route_by_category(state: CustomerServiceState) -> str: return state["category"]# 构建图workflow = StateGraph(CustomerServiceState)workflow.add_node("classify", classify_issue)workflow.add_node("technical", handle_technical)workflow.add_node("refund", handle_refund)workflow.add_node("general", handle_general)workflow.add_edge(START, "classify")workflow.add_conditional_edges("classify", route_by_category, { "technical": "technical", "refund": "refund", "general": "general"})workflow.add_edge("technical", END)workflow.add_edge("refund", END)workflow.add_edge("general", END)app = workflow.compile()
2️⃣ 案例二：带记忆的智能对话机器人
核心功能
- • 多轮对话记忆（使用 MessagesState）
关键代码
💡 MessagesState 和 add_messages 已在 04_Reducer状态管理 中介绍
import osfrom langgraph.graph import StateGraph, START, END, MessagesStatefrom langchain_core.messages import HumanMessage, AIMessage, SystemMessagefrom langchain_community.llms import Tongyios.environ["DASHSCOPE_API_KEY"] = "your-api-key"llm = Tongyi(model="qwen-plus", temperature=0.7, streaming=True)class ChatState(MessagesState): user_name: str conversation_count: intdef chatbot_node(state: ChatState): system_prompt = f"""你是一个友好、专业的AI助手。用户名: {state.get('user_name', '未知用户')}当前是第 {state.get('conversation_count', 0)} 轮对话。""" messages = [SystemMessage(content=system_prompt)] + state["messages"] response = llm.invoke(messages) return { "messages": [AIMessage(content=response)], "conversation_count": state.get("conversation_count", 0) + 1 }workflow = StateGraph(ChatState)workflow.add_node("chatbot", chatbot_node)workflow.add_edge(START, "chatbot")workflow.add_edge("chatbot", END)app = workflow.compile()
关键点
- 1. MessagesState：自动管理消息历史（使用 add_messages Reducer）
- 2. SystemMessage：设置 AI 的角色和行为
3️⃣ 案例三：多智能体协作
系统架构
用户问题 → [路由器Agent] → [技术专家/销售专家/客服专家] → 回复
核心代码
from langchain_community.llms import Tongyirouter_llm = Tongyi(model="qwen-turbo", temperature=0.3)tech_llm = Tongyi(model="qwen-plus", temperature=0.7)sales_llm = Tongyi(model="qwen-plus", temperature=0.8)class TeamState(MessagesState): category: strdef router_node(state: TeamState): user_question = state["messages"][-1].content prompt = f"判断问题类型(technical/sales/support): {user_question}" category = router_llm.invoke(prompt).strip().lower() return {"category": category}def technical_expert_node(state: TeamState): response = tech_llm.invoke(state["messages"][-1].content) return {"messages": [AIMessage(content=f"【技术支持】\n{response}")]}def sales_expert_node(state: TeamState): response = sales_llm.invoke(state["messages"][-1].content) return {"messages": [AIMessage(content=f"【销售顾问】\n{response}")]}def route_to_expert(state: TeamState) -> str: return state["category"]workflow = StateGraph(TeamState)workflow.add_node("router", router_node)workflow.add_node("technical", technical_expert_node)workflow.add_node("sales", sales_expert_node)workflow.add_edge(START, "router")workflow.add_conditional_edges("router", route_to_expert, { "technical": "technical", "sales": "sales", "support": "sales"})workflow.add_edge("technical", END)workflow.add_edge("sales", END)
4️⃣ 案例四：智能编程助手
系统架构
用户需求 → [任务分析] → [代码生成] → [代码审查] → {通过/改进} → END ↓ [代码优化] → 回到审查
核心功能
关键代码
class CodingState(MessagesState): task_type: Literal["generate", "debug", "explain", "optimize"] code: str review_result: str iteration: Annotated[int, operator.add] max_iterations: intdef analyze_task_node(state: CodingState): prompt = f"分析编程请求类型(generate/debug/explain/optimize): {state['messages'][-1].content}" return {"task_type": analyzer_llm.invoke(prompt).strip().lower()}def code_generation_node(state: CodingState): code = coder_llm.invoke(state['messages'][-1].content) return {"code": code, "messages": [AIMessage(content=f"```python\n{code}\n```")]}def code_review_node(state: CodingState): return {"review_result": reviewer_llm.invoke(f"审查代码:\n{state['code']}")}def check_review_result(state: CodingState) -> str: if "需要改进" in state["review_result"] and state["iteration"] < state["max_iterations"]: return "optimize" return "done"# 构建图（带循环）workflow.add_edge("generate", "review")workflow.add_conditional_edges("review", check_review_result, { "optimize": "optimize", "done": END})workflow.add_edge("optimize", "review") # 循环回审查
5️⃣ 生产级应用要点
1. 错误处理
def safe_node(state: MyState): try: # 业务逻辑 result = do_something(state) return result except Exception as e: # 记录错误 print(f"Error: {e}") # 返回错误信息 return {"error": str(e)}
2. 日志记录
import logginglogging.basicConfig(level=logging.INFO)logger = logging.getLogger(__name__)def logged_node(state: MyState): logger.info(f"Processing state: {state}") result = process(state) logger.info(f"Result: {result}") return result
3. 性能监控
import timedef timed_node(state: MyState): start = time.time() result = process(state) elapsed = time.time() - start print(f"Node took {elapsed:.2f}s") return result
🎯 小结
