LangGraph + AI 後端實戰：構建智能客服工單處理系統

 1from typing import Annotated, Literal
 2from dataclasses import dataclass, field
 3from datetime import datetime
 4
 5@dataclass
 6class TicketState:
 7    """工單狀態模型"""
 8    # 基本信息
 9    ticket_id: str
10    user_id: str
11    created_at: datetime
12    subject: str
13    description: str
14    
15    # 分類結果
16    category: Literal[
17        "technical", "billing", "account", 
18        "feature_request", "bug_report", "other"
19    ] = None
20    category_confidence: float = 0.0
21    
22    # 優先級
23    priority: Literal["low", "medium", "high", "critical"] = "medium"
24    urgency_score: float = 0.0
25    
26    # 路由信息
27    assigned_department: str = None
28    assigned_agent_id: str = None
29    
30    # 對話歷史
31    messages: list[dict] = field(default_factory=list)
32    
33    # 回應信息
34    response: str = None
35    response_quality_score: float = 0.0
36    
37    # 狀態追蹤
38    status: str = "open"
39    current_node: str = None
40    processing_history: list[dict] = field(default_factory=list)
41    
42    # 錯誤處理
43    errors: list[str] = field(default_factory=list)
44    retry_count: int = 0
45    max_retries: int = 3

  1from langchain_anthropic import ChatAnthropic
  2from langchain_core.prompts import ChatPromptTemplate
  3from langgraph.graph import StateGraph, END
  4import json
  5import os
  6
  7class TicketAgents:
  8    def __init__(self):
  9        # 模型 ID 從環境變數讀取：請填目前可用的 Claude 模型 ID（見 Anthropic 模型文件），
 10        # 不要寫死已退役的 claude-3-5-sonnet-20241022。
 11        self.model = ChatAnthropic(model=os.environ["LLM_MODEL"])
 12    
 13    # Agent 1: 分類 Agent
 14    def classify_agent(self, state: TicketState) -> TicketState:
 15        """識別工單類型和關鍵信息"""
 16        
 17        prompt = ChatPromptTemplate.from_template("""
 18        分析以下客服工單，提取關鍵信息和分類。
 19        
 20        工單主題：{subject}
 21        工單描述：{description}
 22        
 23        請以 JSON 格式返回：
 24        {{
 25            "category": "technical|billing|account|feature_request|bug_report|other",
 26            "confidence": 0.0-1.0,
 27            "key_issues": ["issue1", "issue2"],
 28            "urgency_signals": ["signal1", "signal2"],
 29            "extraction": {{
 30                "affected_product": "...",
 31                "error_message": "...",
 32                "account_related": true/false,
 33                "payment_related": true/false
 34            }}
 35        }}
 36        """)
 37        
 38        chain = prompt | self.model
 39        response = chain.invoke({
 40            "subject": state.subject,
 41            "description": state.description
 42        })
 43        
 44        # 解析回應
 45        result = json.loads(response.content)
 46        state.category = result["category"]
 47        state.category_confidence = result["confidence"]
 48        state.current_node = "classify"
 49        
 50        # 記錄処理歷史
 51        state.processing_history.append({
 52            "timestamp": datetime.now().isoformat(),
 53            "node": "classify",
 54            "result": result
 55        })
 56        
 57        return state
 58    
 59    # Agent 2: 優先級評估 Agent
 60    def priority_agent(self, state: TicketState) -> TicketState:
 61        """評估工單的優先級"""
 62        
 63        prompt = ChatPromptTemplate.from_template("""
 64        基於以下信息評估工單的優先級：
 65        
 66        類別：{category}
 67        主題：{subject}
 68        描述：{description}
 69        
 70        評估因素：
 71        1. 系統中斷？(critical)
 72        2. 用戶無法訪問？(high)
 73        3. 功能受限？(medium)
 74        4. 功能請求/建議？(low)
 75        
 76        返回 JSON：
 77        {{
 78            "priority": "low|medium|high|critical",
 79            "urgency_score": 0.0-1.0,
 80            "reasoning": "...",
 81            "sla_hours": 24|12|4|1
 82        }}
 83        """)
 84        
 85        chain = prompt | self.model
 86        response = chain.invoke({
 87            "category": state.category,
 88            "subject": state.subject,
 89            "description": state.description
 90        })
 91        
 92        result = json.loads(response.content)
 93        state.priority = result["priority"]
 94        state.urgency_score = result["urgency_score"]
 95        state.current_node = "priority"
 96        
 97        state.processing_history.append({
 98            "timestamp": datetime.now().isoformat(),
 99            "node": "priority",
100            "result": result
101        })
102        
103        return state
104    
105    # Agent 3: 路由 Agent
106    def routing_agent(self, state: TicketState) -> TicketState:
107        """決定工單應該分配給哪個部門"""
108        
109        # 部門配置
110        departments = {
111            "technical": {
112                "name": "技術支持部",
113                "skills": ["api", "integration", "performance"],
114                "avg_response_time": 2,
115                "available_agents": 5
116            },
117            "billing": {
118                "name": "計費部",
119                "skills": ["invoicing", "payment", "refund"],
120                "avg_response_time": 4,
121                "available_agents": 3
122            },
123            "account": {
124                "name": "賬戶管理部",
125                "skills": ["login", "profile", "security"],
126                "avg_response_time": 1,
127                "available_agents": 4
128            }
129        }
130        
131        prompt = ChatPromptTemplate.from_template("""
132        根據工單的類別和優先級，決定分配給哪個部門。
133        
134        類別：{category}
135        優先級：{priority}
136        部門信息：{departments_info}
137        
138        返回 JSON：
139        {{
140            "assigned_department": "technical|billing|account",
141            "reasoning": "...",
142            "estimated_wait_time": "X minutes",
143            "fallback_department": "..."
144        }}
145        """)
146        
147        chain = prompt | self.model
148        response = chain.invoke({
149            "category": state.category,
150            "priority": state.priority,
151            "departments_info": json.dumps(departments, ensure_ascii=False)
152        })
153        
154        result = json.loads(response.content)
155        state.assigned_department = result["assigned_department"]
156        state.current_node = "routing"
157        
158        state.processing_history.append({
159            "timestamp": datetime.now().isoformat(),
160            "node": "routing",
161            "result": result
162        })
163        
164        return state
165    
166    # Agent 4: 回應生成 Agent
167    def response_agent(self, state: TicketState) -> TicketState:
168        """根據工單類型生成初始回覆"""
169        
170        prompt = ChatPromptTemplate.from_template("""
171        根據工單信息生成專業的初始回覆。回覆應該：
172        1. 感謝用戶報告問題
173        2. 確認已收到工單
174        3. 解釋接下來的步驟
175        4. 如果可能，提供初步解決方案
176        
177        類別：{category}
178        優先級：{priority}
179        主題：{subject}
180        描述：{description}
181        部門：{department}
182        
183        生成專業、友好的客服回覆（中文，150-300 字）：
184        """)
185        
186        chain = prompt | self.model
187        response = chain.invoke({
188            "category": state.category,
189            "priority": state.priority,
190            "subject": state.subject,
191            "description": state.description,
192            "department": state.assigned_department
193        })
194        
195        state.response = response.content
196        state.current_node = "response_generation"
197        
198        state.processing_history.append({
199            "timestamp": datetime.now().isoformat(),
200            "node": "response_generation",
201            "response": state.response
202        })
203        
204        return state
205    
206    # Agent 5: 品質檢查 Agent
207    def quality_check_agent(self, state: TicketState) -> TicketState:
208        """驗證回應品質"""
209        
210        prompt = ChatPromptTemplate.from_template("""
211        評估以下客服回覆的品質：
212        
213        工單：{subject}
214        回覆：{response}
215        
216        評估維度：
217        1. 專業性 (0-1)
218        2. 清晰性 (0-1)
219        3. 完整性 (0-1)
220        4. 友好性 (0-1)
221        5. 相關性 (0-1)
222        
223        返回 JSON：
224        {{
225            "overall_score": 0.0-1.0,
226            "scores": {{
227                "professionalism": 0.0-1.0,
228                "clarity": 0.0-1.0,
229                "completeness": 0.0-1.0,
230                "friendliness": 0.0-1.0,
231                "relevance": 0.0-1.0
232            }},
233            "issues": ["issue1", "issue2"],
234            "recommendations": ["fix1", "fix2"],
235            "approved": true/false
236        }}
237        """)
238        
239        chain = prompt | self.model
240        response = chain.invoke({
241            "subject": state.subject,
242            "response": state.response
243        })
244        
245        result = json.loads(response.content)
246        state.response_quality_score = result["overall_score"]
247        state.current_node = "quality_check"
248        
249        # 如果品質不符合要求，生成改進建議
250        if not result["approved"]:
251            state.errors.append(f"品質檢查未通過：{result['issues']}")
252            return state
253        
254        state.processing_history.append({
255            "timestamp": datetime.now().isoformat(),
256            "node": "quality_check",
257            "score": state.response_quality_score,
258            "approved": True
259        })
260        
261        return state

 1from langgraph.graph import StateGraph, START, END
 2
 3def build_ticket_workflow():
 4    """構建工單處理工作流"""
 5    
 6    workflow = StateGraph(TicketState)
 7    agents = TicketAgents()
 8    
 9    # 添加節點
10    workflow.add_node("classify", agents.classify_agent)
11    workflow.add_node("priority", agents.priority_agent)
12    workflow.add_node("routing", agents.routing_agent)
13    workflow.add_node("response_generation", agents.response_agent)
14    workflow.add_node("quality_check", agents.quality_check_agent)
15    
16    # 添加邊（節點間的轉移）
17    workflow.add_edge(START, "classify")
18    workflow.add_edge("classify", "priority")
19    workflow.add_edge("priority", "routing")
20    workflow.add_edge("routing", "response_generation")
21    workflow.add_edge("response_generation", "quality_check")
22    
23    # 品質檢查的條件轉移
24    def check_quality(state: TicketState):
25        if state.response_quality_score >= 0.8:
26            return "end"
27        else:
28            return "response_generation"  # 重新生成
29    
30    workflow.add_conditional_edges(
31        "quality_check",
32        check_quality,
33        {
34            "end": END,
35            "response_generation": "response_generation"
36        }
37    )
38    
39    return workflow.compile()

 1# 初始化工作流
 2ticket_processor = build_ticket_workflow()
 3
 4# 創建新工單
 5new_ticket = TicketState(
 6    ticket_id="TKT-2024-001",
 7    user_id="user_123",
 8    created_at=datetime.now(),
 9    subject="API 認證失敗導致集成中斷",
10    description="""
11    我們的應用無法連接到您的 API。
12    錯誤信息：401 Unauthorized
13    這發生在上午 10:00 UTC，影響了所有用戶。
14    """
15)
16
17# 執行工作流
18print("開始處理工單...")
19result = ticket_processor.invoke(new_ticket)
20
21# 查看結果
22print(f"工單 ID: {result.ticket_id}")
23print(f"分類: {result.category} (置信度: {result.category_confidence:.2%})")
24print(f"優先級: {result.priority}")
25print(f"分配部門: {result.assigned_department}")
26print(f"回應品質分數: {result.response_quality_score:.2f}")
27print(f"生成的回覆:\n{result.response}")
28print(f"\n処理歷史:")
29for entry in result.processing_history:
30    print(f"  - {entry['node']}: {entry['timestamp']}")

 1def conversation_agent(self, state: TicketState) -> TicketState:
 2    """支持多輪對話的 Agent"""
 3    
 4    # 構建對話歷史
 5    history = "\n".join([
 6        f"{msg['role']}: {msg['content']}" 
 7        for msg in state.messages[-5:]  # 最近 5 條消息
 8    ])
 9    
10    prompt = ChatPromptTemplate.from_template("""
11    你是一個客服代理。基於以下工單歷史和對話，提供幫助。
12    
13    工單背景：
14    - 類別：{category}
15    - 優先級：{priority}
16    
17    對話歷史：
18    {history}
19    
20    用戶最新消息：{user_message}
21    
22    生成有幫助的回覆：
23    """)
24    
25    chain = prompt | self.model
26    response = chain.invoke({
27        "category": state.category,
28        "priority": state.priority,
29        "history": history,
30        "user_message": state.messages[-1]["content"]
31    })
32    
33    # 添加到消息歷史
34    state.messages.append({
35        "role": "assistant",
36        "content": response.content,
37        "timestamp": datetime.now().isoformat()
38    })
39    
40    return state

 1def add_error_handling(workflow):
 2    """為工作流添加錯誤處理機制"""
 3    
 4    def handle_error(state: TicketState, error: Exception):
 5        state.errors.append(str(error))
 6        state.retry_count += 1
 7        
 8        if state.retry_count >= state.max_retries:
 9            state.status = "escalated"
10            # 升級給人工客服
11            return "escalate_to_human"
12        else:
13            # 重試
14            return f"retry_{state.current_node}"
15    
16    return workflow

 1def monitor_workflow(state: TicketState):
 2    """實時監控工作流狀態"""
 3    
 4    metrics = {
 5        "ticket_id": state.ticket_id,
 6        "status": state.status,
 7        "processing_time": calculate_processing_time(state),
 8        "quality_score": state.response_quality_score,
 9        "retry_count": state.retry_count,
10        "error_count": len(state.errors)
11    }
12    
13    # 發送到監控系統
14    send_to_monitoring_system(metrics)
15    
16    # 告警規則
17    if state.retry_count >= 2:
18        alert(f"工單 {state.ticket_id} 重試次數過多")
19    
20    if state.response_quality_score < 0.7:
21        alert(f"工單 {state.ticket_id} 回應品質低")

 1FROM python:3.11-slim
 2
 3WORKDIR /app
 4
 5COPY requirements.txt .
 6RUN pip install -r requirements.txt
 7
 8COPY . .
 9
10EXPOSE 8000
11
12CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

 1from fastapi import FastAPI, BackgroundTasks
 2from pydantic import BaseModel
 3
 4app = FastAPI()
 5ticket_processor = build_ticket_workflow()
 6
 7class TicketRequest(BaseModel):
 8    user_id: str
 9    subject: str
10    description: str
11
12@app.post("/tickets")
13async def create_ticket(request: TicketRequest, background_tasks: BackgroundTasks):
14    """創建新工單"""
15    
16    ticket = TicketState(
17        ticket_id=generate_ticket_id(),
18        user_id=request.user_id,
19        created_at=datetime.now(),
20        subject=request.subject,
21        description=request.description
22    )
23    
24    # 後台處理工單
25    background_tasks.add_task(process_ticket, ticket)
26    
27    return {"ticket_id": ticket.ticket_id, "status": "accepted"}
28
29async def process_ticket(ticket: TicketState):
30    """處理工單"""
31    result = ticket_processor.invoke(ticket)
32    save_to_database(result)
33    notify_user(result)