LangGraph 最強大的地方不是它能做什麼，而是它開啟了什麼。本文探討 10 個生產級的創意應用案例，展示如何用 LangGraph 構建明天的 AI 系統。
1. 智能內容審核和分類系統
應用場景
社交媒體平台、論壇、UGC 社區都面臨內容審核的挑戰。傳統方案要麼依賴人工，要麼規則過於死板。LangGraph 能構建多層次的智能審核系統。
工作流設計
用戶上傳內容
↓
[快速過濾] - 明顯違規內容
│
├→ 違規 → 直接刪除
│
└→ 可疑 → 深度分析
↓
[語境分析] - 理解上下文
├→ 色情內容識別
├→ 暴力內容識別
├→ 仇恨言論檢測
└→ 垃圾廣告檢測
↓
[優先級評分] - 確定處理優先級
↓
[人工審核隊列] - 將爭議內容分配給人工
↓
[持續學習] - 從人工審核結果改進模型
核心實現
1from dataclasses import dataclass
2from enum import Enum
3from langgraph.graph import StateGraph, START, END
4
5class ContentRisk(str, Enum):
6 SAFE = "safe"
7 WARN = "warn"
8 REMOVE = "remove"
9 REVIEW = "review"
10
11@dataclass
12class ContentModerationState:
13 content_id: str
14 content_text: str
15 user_id: str
16
17 # 審核結果
18 risk_level: ContentRisk = None
19 risk_score: float = 0.0 # 0-1
20 risk_categories: list[str] = None
21 context_analysis: str = None
22 human_review_needed: bool = False
23 confidence: float = 0.0
24
25class ContentModerationWorkflow:
26 def rapid_filter(self, state: ContentModerationState) -> ContentModerationState:
27 """快速過濾明顯違規內容"""
28 illegal_keywords = ["枪", "炸弹", "毒药"] # 簡化示例
29
30 if any(kw in state.content_text for kw in illegal_keywords):
31 state.risk_level = ContentRisk.REMOVE
32 state.risk_score = 0.99
33 return state
34
35 # 進入深度分析
36 return state
37
38 def deep_analysis(self, state: ContentModerationState) -> ContentModerationState:
39 """深度內容分析"""
40 from langchain_anthropic import ChatAnthropic
41
42 model = ChatAnthropic()
43
44 prompt = f"""
45 分析以下內容的風險等級。
46
47 內容：{state.content_text}
48
49 返回 JSON：
50 {{
51 "risk_categories": ["色情", "暴力", "垃圾", "仇恨", "其他"],
52 "risk_score": 0.0-1.0,
53 "context": "簡短說明",
54 "requires_human_review": true/false
55 }}
56 """
57
58 response = model.invoke(prompt)
59 result = eval(response.content)
60
61 state.risk_score = result["risk_score"]
62 state.risk_categories = result["risk_categories"]
63 state.context_analysis = result["context"]
64 state.human_review_needed = result["requires_human_review"]
65 state.confidence = 0.95 # 基於模型配置
66
67 return state
68
69 def decide_action(self, state: ContentModerationState) -> str:
70 """決定審核動作"""
71 if state.risk_score >= 0.9:
72 return "remove"
73 elif state.risk_score >= 0.7:
74 return "warn"
75 elif state.human_review_needed:
76 return "review"
77 else:
78 return "approve"
79
80 def build_workflow(self):
81 workflow = StateGraph(ContentModerationState)
82
83 workflow.add_node("rapid_filter", self.rapid_filter)
84 workflow.add_node("deep_analysis", self.deep_analysis)
85
86 workflow.add_edge(START, "rapid_filter")
87
88 def should_analyze(state):
89 return state.risk_level is None # 如果已判定為直接刪除，不分析
90
91 workflow.add_conditional_edges(
92 "rapid_filter",
93 lambda s: "deep_analysis" if s.risk_level is None else "decide",
94 {"deep_analysis": "deep_analysis"}
95 )
96
97 workflow.add_node("decide", lambda s: s)
98 workflow.add_edge("deep_analysis", "decide")
99 workflow.add_edge("decide", END)
100
101 return workflow.compile()
2. 實時數據分析和報告生成系統
應用場景
金融機構需要即時分析市場數據並生成報告，數據分析平台需要自動洞察和警報。
工作流設計
數據流入
↓
[異常檢測] - 識別異常數據點
↓
[特徵提取] - 提取關鍵特徵
↓
[因果分析] - AI 分析原因
├→ 是否預示趨勢變化？
├→ 與歷史對比如何？
└→ 相關外部事件？
↓
[報告生成] - 生成自然語言報告
↓
[優先級評分] - 決定是否立即告警
↓
[多渠道分發] - 郵件、Slack、儀表板
核心實現
1@dataclass
2class DataAnalysisState:
3 data_point: dict # {"metric": "CPU", "value": 95, "timestamp": ...}
4 historical_data: list[dict]
5 analysis_result: dict = None
6 confidence: float = 0.0
7 requires_alert: bool = False
8
9class DataAnalysisAgent:
10 def anomaly_detection(self, state: DataAnalysisState) -> DataAnalysisState:
11 """異常檢測"""
12 import statistics
13
14 values = [d["value"] for d in state.historical_data[-100:]]
15 mean = statistics.mean(values)
16 stdev = statistics.stdev(values) if len(values) > 1 else 0
17
18 # 如果超過 3 個標準差
19 z_score = (state.data_point["value"] - mean) / stdev if stdev > 0 else 0
20
21 state.is_anomaly = abs(z_score) > 3
22 state.z_score = z_score
23
24 return state
25
26 def ai_root_cause_analysis(self, state: DataAnalysisState) -> DataAnalysisState:
27 """AI 根因分析"""
28 if not state.is_anomaly:
29 return state
30
31 from langchain_anthropic import ChatAnthropic
32
33 model = ChatAnthropic()
34
35 prompt = f"""
36 分析以下異常的根本原因：
37
38 指標：{state.data_point['metric']}
39 當前值：{state.data_point['value']}
40 歷史均值：{statistics.mean([d['value'] for d in state.historical_data[-100:]])}
41 時間戳：{state.data_point['timestamp']}
42
43 可能的原因和建議的行動：
44 """
45
46 response = model.invoke(prompt)
47 state.analysis_result = {
48 "root_cause": response.content,
49 "confidence": 0.85
50 }
51
52 return state
53
54 def decide_alert(self, state: DataAnalysisState) -> str:
55 """決定是否告警"""
56 if state.is_anomaly and state.z_score > 5:
57 return "critical_alert"
58 elif state.is_anomaly:
59 return "warning_alert"
60 else:
61 return "no_alert"
3. 個性化教育內容生成系統
應用場景
在線教育平台、企業培訓、個性化學習路徑。
工作流設計
學生提交問題
↓
[理解能力評估] - 評估學生當前水平
↓
[內容匹配] - 找到合適難度的解釋
├→ 太簡單？提高複雜度
├→ 太難？簡化概念
└→ 剛好？深化理解
↓
[多角度解釋] - 用不同方式解釋
├→ 代碼示例
├→ 類比說明
├→ 視覺圖表
└→ 實際應用案例
↓
[互動練習生成] - 創建個性化練習
↓
[反饋和調整] - 根據表現調整策略
核心實現
1@dataclass
2class StudentLearningState:
3 student_id: str
4 current_question: str
5 learning_history: list[dict] # [{"topic": "...", "score": 0.8}]
6 proficiency_level: float # 0-1
7
8 # 生成的內容
9 explanation: str = None
10 examples: list[str] = None
11 exercises: list[dict] = None
12
13class AdaptiveLearningAgent:
14 def assess_proficiency(self, state: StudentLearningState) -> StudentLearningState:
15 """評估學生水平"""
16 if not state.learning_history:
17 state.proficiency_level = 0.5 # 新用戶
18 return state
19
20 # 基於過往記錄計算水平
21 recent_scores = [h["score"] for h in state.learning_history[-10:]]
22 state.proficiency_level = sum(recent_scores) / len(recent_scores)
23
24 return state
25
26 def generate_explanation(self, state: StudentLearningState) -> StudentLearningState:
27 """生成個性化解釋"""
28 from langchain_anthropic import ChatAnthropic
29
30 model = ChatAnthropic()
31
32 # 根據水平選擇難度
33 difficulty = "simple" if state.proficiency_level < 0.3 else \
34 "medium" if state.proficiency_level < 0.7 else "advanced"
35
36 prompt = f"""
37 用{difficulty}難度解釋以下概念。
38
39 問題：{state.current_question}
40
41 包含：
42 1. 簡潔定義
43 2. 實際例子
44 3. 常見誤區
45 """
46
47 response = model.invoke(prompt)
48 state.explanation = response.content
49
50 return state
51
52 def generate_practice(self, state: StudentLearningState) -> StudentLearningState:
53 """生成個性化練習"""
54 from langchain_anthropic import ChatAnthropic
55
56 model = ChatAnthropic()
57
58 prompt = f"""
59 為水平為 {state.proficiency_level:.1%} 的學生生成 3 道練習題。
60
61 主題：{state.current_question}
62
63 返回 JSON：
64 [
65 {{"question": "...", "options": [...], "correct_answer": "..."}},
66 ...
67 ]
68 """
69
70 response = model.invoke(prompt)
71 import json
72 state.exercises = json.loads(response.content)
73
74 return state
4. 智能代碼審查和優化系統
應用場景
GitHub Actions、GitLab CI/CD、IDE 插件。開發者推送代碼時自動審查。
工作流設計
代碼變更
↓
[靜態分析] - 檢查常見問題
├→ 安全漏洞
├→ 性能問題
└→ 代碼風格
↓
[AI 代碼審查] - 深度代碼理解
├→ 邏輯是否正確？
├→ 邊界情況是否處理？
├→ 是否有更好的實現？
└→ 文檔是否充分？
↓
[測試覆蓋分析] - 是否有必要的測試？
↓
[自動優化建議] - 性能和可讀性優化
↓
[報告生成] - 生成 PR 審查注釋
核心實現
1@dataclass
2class CodeReviewState:
3 pr_id: str
4 files_changed: list[dict] # [{"filename": "...", "diff": "..."}]
5 security_issues: list[dict] = None
6 performance_issues: list[dict] = None
7 suggestions: list[str] = None
8 test_coverage: float = 0.0
9
10class CodeReviewAgent:
11 def analyze_security(self, state: CodeReviewState) -> CodeReviewState:
12 """安全性分析"""
13 from langchain_anthropic import ChatAnthropic
14
15 model = ChatAnthropic()
16
17 all_code = "\n\n".join([
18 f"# {f['filename']}\n{f['diff']}"
19 for f in state.files_changed
20 ])
21
22 prompt = f"""
23 分析以下代碼的安全問題。
24
25 代碼：
26 {all_code}
27
28 返回 JSON：
29 {{
30 "issues": [
31 {{"type": "SQL Injection", "severity": "high", "location": "line X"}},
32 ...
33 ],
34 "recommendations": ["..."]
35 }}
36 """
37
38 response = model.invoke(prompt)
39 import json
40 result = json.loads(response.content)
41 state.security_issues = result["issues"]
42
43 return state
44
45 def suggest_improvements(self, state: CodeReviewState) -> CodeReviewState:
46 """建議改進"""
47 from langchain_anthropic import ChatAnthropic
48
49 model = ChatAnthropic()
50
51 prompt = f"""
52 建議代碼改進（性能、可讀性、最佳實踐）。
53
54 當前代碼缺陷：
55 - 安全問題：{len(state.security_issues or [])} 個
56 - 性能問題：{len(state.performance_issues or [])} 個
57
58 建議優化的 TOP 3：
59 """
60
61 response = model.invoke(prompt)
62 state.suggestions = response.content.split("\n")
63
64 return state
5. 多語言客服系統
應用場景
全球化業務，客戶來自不同國家和語言背景。
工作流設計
用戶提交工單（任意語言）
↓
[語言檢測和翻譯]
├→ 檢測源語言
├→ 翻譯為工作語言（英文）
└→ 保存原始語言
↓
[智能分類和路由]
├→ 識別問題類型
├→ 匹配最適合的服務代表（可能需要語言技能）
└→ 評估優先級
↓
[生成回應（原始語言）]
└→ 在源語言中回覆
↓
[質量檢查]
├→ 確保翻譯準確
└→ 驗證上下文適當性
核心實現
1from langdetect import detect
2from langchain.chains import LLMChain
3
4@dataclass
5class MultilingualTicketState:
6 ticket_id: str
7 original_text: str
8 source_language: str = None
9 working_language_text: str = None # 翻譯為英文
10 category: str = None
11 response_draft: str = None
12 response_final: str = None # 翻譯回源語言
13
14class MultilingualAgent:
15 async def detect_and_translate(self, state: MultilingualTicketState) -> MultilingualTicketState:
16 """檢測語言並翻譯"""
17 from langchain_anthropic import ChatAnthropic
18
19 # 檢測語言
20 state.source_language = detect(state.original_text)
21
22 if state.source_language == "en":
23 state.working_language_text = state.original_text
24 else:
25 # 翻譯
26 model = ChatAnthropic()
27 prompt = f"""
28 將以下 {state.source_language} 文本翻譯為英文。
29
30 文本：{state.original_text}
31
32 只返回翻譯結果，不要其他說明。
33 """
34
35 response = model.invoke(prompt)
36 state.working_language_text = response.content
37
38 return state
39
40 async def generate_response_in_target_language(self, state: MultilingualTicketState) -> MultilingualTicketState:
41 """用源語言生成回應"""
42 from langchain_anthropic import ChatAnthropic
43
44 model = ChatAnthropic()
45
46 # 先用英文生成
47 prompt = f"""
48 根據以下工單信息生成客服回應。
49
50 工單：{state.working_language_text}
51 類別：{state.category}
52 """
53
54 response = model.invoke(prompt)
55 state.response_draft = response.content
56
57 # 如果不是英文，翻譯回源語言
58 if state.source_language != "en":
59 translate_prompt = f"""
60 將以下英文回應翻譯為 {state.source_language}，保持專業且友好的語氣。
61
62 英文：{state.response_draft}
63 """
64
65 trans_response = model.invoke(translate_prompt)
66 state.response_final = trans_response.content
67 else:
68 state.response_final = state.response_draft
69
70 return state
6. 實時知識圖譜構建系統
應用場景
文檔管理、企業知識庫、自動化 Wiki 維護。
工作流設計
新文檔上傳
↓
[概念提取] - 識別關鍵概念和實體
↓
[關係發現] - 發現概念間的關係
├→ 包含關係（is-a）
├→ 因果關係（causes）
└→ 相似關係（similar-to）
↓
[衝突檢測] - 檢查是否與現有知識衝突
↓
[圖譜更新] - 更新知識圖
↓
[自動交叉鏈接] - 連結相關文檔
核心實現
1@dataclass
2class KnowledgeGraphState:
3 document_id: str
4 content: str
5 extracted_entities: list[str] = None
6 extracted_relations: list[tuple] = None # [(entity1, relation, entity2)]
7 conflicts: list[dict] = None
8 new_links: list[dict] = None
9
10class KnowledgeGraphAgent:
11 def extract_entities(self, state: KnowledgeGraphState) -> KnowledgeGraphState:
12 """提取實體"""
13 from langchain_anthropic import ChatAnthropic
14
15 model = ChatAnthropic()
16
17 prompt = f"""
18 從以下文本中提取所有關鍵概念和實體。
19
20 文本：{state.content[:1000]}...
21
22 返回 JSON：
23 ["entity1", "entity2", "entity3", ...]
24 """
25
26 response = model.invoke(prompt)
27 import json
28 state.extracted_entities = json.loads(response.content)
29
30 return state
31
32 def extract_relations(self, state: KnowledgeGraphState) -> KnowledgeGraphState:
33 """提取關係"""
34 from langchain_anthropic import ChatAnthropic
35
36 model = ChatAnthropic()
37
38 entities_str = ", ".join(state.extracted_entities)
39
40 prompt = f"""
41 識別以下實體之間的關係。
42
43 實體：{entities_str}
44 文本：{state.content[:1000]}...
45
46 返回 JSON：
47 [
48 {{"entity1": "A", "relation": "causes", "entity2": "B"}},
49 ...
50 ]
51 """
52
53 response = model.invoke(prompt)
54 import json
55 state.extracted_relations = json.loads(response.content)
56
57 return state
7. AI 驅動的故障排查系統
應用場景
IT 運維、應用診斷、客戶支持。
工作流設計
故障報告
↓
[症狀分析] - 分析報錯消息和日誌
├→ 解析堆棧跟踪
├→ 識別異常模式
└→ 提取關鍵信息
↓
[根因推理] - AI 推測根本原因
├→ 可能原因排序
├→ 置信度評分
└→ 所需更多信息
↓
[自動解決步驟] - 如果可能的話自動修復
├→ 清理緩存？
├→ 重啟服務？
├→ 更新配置？
└→ 需要人工介入？
↓
[驗證] - 確認問題解決
↓
[知識存儲] - 存儲此案例供未來參考
核心實現
1@dataclass
2class TroubleshootingState:
3 ticket_id: str
4 error_message: str
5 logs: str
6 system_info: dict
7
8 suspected_causes: list[dict] = None # [{"cause": "...", "confidence": 0.8}]
9 suggested_fixes: list[str] = None
10 auto_resolution_attempted: bool = False
11 resolution_successful: bool = False
12
13class TroubleshootingAgent:
14 def analyze_symptoms(self, state: TroubleshootingState) -> TroubleshootingState:
15 """分析症狀"""
16 from langchain_anthropic import ChatAnthropic
17
18 model = ChatAnthropic()
19
20 prompt = f"""
21 分析以下故障症狀。
22
23 錯誤消息：{state.error_message}
24 日誌摘錄：{state.logs[:500]}...
25 系統信息：{state.system_info}
26
27 可能的根本原因（按可能性排序）：
28 """
29
30 response = model.invoke(prompt)
31
32 # 解析響應提取原因
33 causes = response.content.split("\n")
34 state.suspected_causes = [
35 {"cause": cause.strip(), "confidence": 0.8 - i*0.1}
36 for i, cause in enumerate(causes[:5])
37 ]
38
39 return state
40
41 def suggest_fixes(self, state: TroubleshootingState) -> TroubleshootingState:
42 """建議修復步驟"""
43 from langchain_anthropic import ChatAnthropic
44
45 model = ChatAnthropic()
46
47 top_cause = state.suspected_causes[0]["cause"]
48
49 prompt = f"""
50 根據根本原因建議修復步驟。
51
52 根本原因：{top_cause}
53
54 返回具體的、可自動執行的步驟：
55 1.
56 2.
57 3.
58 """
59
60 response = model.invoke(prompt)
61 state.suggested_fixes = response.content.split("\n")
62
63 return state
8. 實時合同分析和風險評估系統
應用場景
法律科技、合同管理、合規檢查。
工作流設計
合同上傳
↓
[文本提取和結構化] - 識別合同的各個章節
↓
[條款分析]
├→ 支付條款
├→ 責任條款
├→ 終止條款
├→ 保密協議
└→ 知識產權
↓
[風險識別] - 發現潛在風險
├→ 不利的支付條款
├→ 過度的責任承諾
├→ 不合理的終止條款
└→ 模糊的定義
↓
[對標分析] - 與標準合同對比
↓
[建議修訂] - 生成修訂建議
↓
[優先級排序] - 重要問題優先
核心實現
1@dataclass
2class ContractAnalysisState:
3 contract_id: str
4 contract_text: str
5 contract_type: str # NDA, SLA, Employment, etc.
6
7 sections: dict = None # {"terms": "...", "liability": "..."}
8 risk_items: list[dict] = None
9 estimated_exposure: str = None # low, medium, high, critical
10 recommendations: list[str] = None
11
12class ContractAnalysisAgent:
13 def extract_sections(self, state: ContractAnalysisState) -> ContractAnalysisState:
14 """提取合同章節"""
15 from langchain_anthropic import ChatAnthropic
16
17 model = ChatAnthropic()
18
19 prompt = f"""
20 識別 {state.contract_type} 合同的主要章節。
21
22 合同（前 2000 字）：
23 {state.contract_text[:2000]}...
24
25 返回 JSON：
26 {{
27 "sections": {{
28 "payment_terms": "...",
29 "liability": "...",
30 "termination": "...",
31 "confidentiality": "...",
32 "intellectual_property": "..."
33 }}
34 }}
35 """
36
37 response = model.invoke(prompt)
38 import json
39 state.sections = json.loads(response.content)["sections"]
40
41 return state
42
43 def identify_risks(self, state: ContractAnalysisState) -> ContractAnalysisState:
44 """識別風險"""
45 from langchain_anthropic import ChatAnthropic
46
47 model = ChatAnthropic()
48
49 prompt = f"""
50 識別此合同的法律和商業風險。
51
52 合同類型：{state.contract_type}
53 支付條款：{state.sections.get('payment_terms')}
54 責任：{state.sections.get('liability')}
55
56 列出所有風險項（嚴重性: low/medium/high/critical）：
57
58 返回 JSON：
59 [
60 {{"issue": "支付期限為 120 天", "severity": "high", "recommendation": "爭取改為 30 天"}},
61 ...
62 ]
63 """
64
65 response = model.invoke(prompt)
66 import json
67 state.risk_items = json.loads(response.content)
68
69 # 評估總體風險
70 severities = [item["severity"] for item in state.risk_items]
71 if "critical" in severities:
72 state.estimated_exposure = "critical"
73 elif severities.count("high") >= 2:
74 state.estimated_exposure = "high"
75 else:
76 state.estimated_exposure = "medium"
77
78 return state
9. 實時個人化推薦系統
應用場景
電子商務、流媒體、內容平台。
核心特點
用戶行為流
↓
[實時特徵計算]
├→ 瀏覽歷史
├→ 購買行為
├→ 搜索查詢
└→ 互動信號
↓
[AI 理解用戶意圖]
├→ 用戶想要什麼？
├→ 用戶在尋找什麼類型的產品？
└→ 用戶有什麼痛點？
↓
[實時推薦生成]
├→ 精準匹配
├→ 多樣化
└→ 新鮮度平衡
↓
[A/B 測試]
└→ 持續優化
10. 智能研究助手系統
應用場景
學術研究、市場調研、競爭分析。
工作流設計
研究問題
↓
[文獻搜索] - 查找相關文獻
↓
[內容摘要] - 總結每篇文獻
↓
[共識提取] - 在文獻間尋找共識
↓
[分歧識別] - 識別不同觀點
↓
[見解綜合] - AI 生成新見解
↓
[報告生成] - 構建研究報告
├→ 執行摘要
├→ 背景
├→ 主要發現
├→ 矛盾的觀點
├→ 建議下一步研究
└→ 參考文獻
未來方向：LangGraph 進階應用
1. 自適應工作流
根據中間結果動態調整工作流。
示例：
- 如果分類置信度低，自動增加驗證步驟
- 如果需要人工審核，自動路由給合適的人
- 根據歷史成功率調整 Agent 組合
2. 多 Agent 辯論系統
多個 Agent 對立場進行辯論，AI 仲裁。
優點：
- 更全面的分析
- 減少單一 Agent 的偏見
- 生成更有見地的結論
3. 人機混合工作流
AI 完成 70% 工作，人工審查 30%。
優勢：
- 充分利用 AI 的速度
- 保持人工的判斷力
- 持續學習和改進
4. 多模態工作流
不僅處理文本，還包括：
- 圖像分析
- 音視頻處理
- 表格數據
- 實時流數據
5. 跨組織工作流
多個組織的系統協作。
用例：
- 供應鏈協調
- 多方審批流程
- 去中心化決策
選擇合適的用例
| 場景 | 複雜度 | 價值 | 實現周期 |
|---|---|---|---|
| 客服分類 | 低 | 中 | 1-2 周 |
| 內容審核 | 中 | 高 | 2-4 周 |
| 代碼審查 | 中 | 高 | 3-6 周 |
| 合同分析 | 高 | 很高 | 4-8 周 |
| 知識圖譜 | 高 | 很高 | 6-12 周 |
總結
LangGraph 的威力在於它能將複雜的決策邏輯變成可追蹤、可控制的工作流。每個上述案例都代表著一個真實的商業價值——效率提升、成本降低、質量改善。
選擇一個靠近你的領域的案例，開始你的 LangGraph 之旅。未來屬於那些能夠將 AI 集成到業務流程中的公司。
