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

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