Day 7: 週總結：整合 Gemini 與 LangGraph

經過一週的學習，我們已經掌握了 Gemini API 和 LangGraph 的核心技術。今天讓我們回顧這週的學習成果，並建立一個整合兩者優勢的綜合應用 - 「智能工作助理」！

📚 本週學習回顧

Day 1-3：基礎建立

• ✅ 環境準備：搭建完整的開發環境
• ✅ Gemini API：掌握基本呼叫和進階功能
• ✅ Gemini CLI：熟悉命令列工具的便利操作

Day 4-6：實戰應用

• ✅ 對話機器人：建立具備記憶和指令系統的聊天機器人
• ✅ LangGraph 概念：理解圖形化工作流程的核心思維
• ✅ 學習助手：實作複雜的多節點 AI 應用

🎯 整合專案：智能工作助理

讓我們建立一個整合 Gemini 和 LangGraph 的綜合應用，結合前面學到的所有技術：

專案特色

• 🤖 多模態交互：支援文字、圖片、文件處理
• 🔄 複雜工作流程：使用 LangGraph 管理多步驟任務
• 💼 實用功能：郵件撰寫、會議總結、資料分析等
• 🧠 智能記憶：記住工作偏好和歷史互動

🏗 架構設計

smart_work_assistant/
├── main.py                    # 主程式
├── core/
│   ├── __init__.py
│   ├── state_manager.py       # 狀態管理
│   ├── workflow_engine.py     # 工作流程引擎
│   └── memory_system.py       # 記憶系統
├── integrations/
│   ├── __init__.py
│   ├── gemini_integration.py  # Gemini 整合
│   ├── cli_bridge.py          # CLI 橋接
│   └── file_handler.py        # 文件處理
├── workflows/
│   ├── __init__.py
│   ├── email_workflow.py      # 郵件工作流程
│   ├── meeting_workflow.py    # 會議工作流程
│   ├── analysis_workflow.py   # 分析工作流程
│   └── creative_workflow.py   # 創意工作流程
├── utils/
│   ├── __init__.py
│   ├── prompt_templates.py    # 提示詞模板
│   └── formatters.py          # 格式化工具
└── config/
    ├── __init__.py
    └── settings.py            # 設定檔

🔧 核心整合程式碼

1. 整合狀態管理 (core/state_manager.py)

from typing import TypedDict, List, Dict, Optional, Any
from enum import Enum
from datetime import datetime

class TaskType(Enum):
    EMAIL = "email_composition"
    MEETING = "meeting_summary"
    ANALYSIS = "data_analysis"
    CREATIVE = "creative_writing"
    PLANNING = "task_planning"
    RESEARCH = "research_assistance"

class WorkflowState(TypedDict):
    # 使用者輸入
    user_input: str
    attachments: List[Dict[str, Any]]
    user_preferences: Dict[str, Any]
    
    # 任務識別
    task_type: str
    priority_level: str
    estimated_duration: str
    
    # Gemini 回應
    gemini_analysis: Optional[Dict]
    gemini_response: Optional[str]
    
    # LangGraph 狀態
    current_node: str
    processing_history: List[str]
    intermediate_results: List[Dict]
    
    # 記憶系統
    conversation_context: List[Dict]
    user_profile: Dict[str, Any]
    past_interactions: List[Dict]
    
    # 最終輸出
    final_output: str
    confidence_score: float
    suggestions: List[str]
    next_actions: List[str]

class StateManager:
    def __init__(self):
        self.current_state: WorkflowState = self._initialize_state()
        self.state_history: List[WorkflowState] = []
    
    def _initialize_state(self) -> WorkflowState:
        return {
            "user_input": "",
            "attachments": [],
            "user_preferences": {},
            "task_type": "",
            "priority_level": "medium",
            "estimated_duration": "",
            "gemini_analysis": None,
            "gemini_response": None,
            "current_node": "start",
            "processing_history": [],
            "intermediate_results": [],
            "conversation_context": [],
            "user_profile": {},
            "past_interactions": [],
            "final_output": "",
            "confidence_score": 0.0,
            "suggestions": [],
            "next_actions": []
        }
    
    def update_state(self, updates: Dict[str, Any]) -> WorkflowState:
        """更新狀態並記錄歷史"""
        self.state_history.append(self.current_state.copy())
        
        # 更新當前狀態
        for key, value in updates.items():
            if key in self.current_state:
                self.current_state[key] = value
        
        return self.current_state
    
    def get_context_summary(self) -> str:
        """獲取上下文摘要"""
        return f"""
        當前任務：{self.current_state['task_type']}
        處理階段：{self.current_state['current_node']}
        優先級：{self.current_state['priority_level']}
        處理歷史：{' -> '.join(self.current_state['processing_history'])}
        """

2. Gemini 深度整合 (integrations/gemini_integration.py)

import google.generativeai as genai
import json
from typing import Dict, List, Optional, Any
from core.state_manager import WorkflowState, TaskType
import subprocess
import tempfile
import os

class AdvancedGeminiClient:
    def __init__(self):
        genai.configure(api_key=os.getenv('GEMINI_API_KEY'))
        self.pro_model = genai.GenerativeModel('gemini-pro')
        self.vision_model = genai.GenerativeModel('gemini-pro-vision')
        
    def analyze_work_request(self, state: WorkflowState) -> Dict[str, Any]:
        """深度分析工作請求"""
        user_input = state["user_input"]
        context = state["conversation_context"]
        preferences = state["user_preferences"]
        
        prompt = f"""
        作為專業的工作助理 AI，分析以下工作請求：
        
        請求內容：{user_input}
        歷史對話：{json.dumps(context[-3:], ensure_ascii=False) if context else "無"}
        使用者偏好：{json.dumps(preferences, ensure_ascii=False)}
        
        請以 JSON 格式分析：
        {{
            "task_type": "email_composition | meeting_summary | data_analysis | creative_writing | task_planning | research_assistance",
            "priority_level": "low | medium | high | urgent",
            "key_requirements": ["需求1", "需求2"],
            "estimated_duration": "預估時間",
            "required_skills": ["技能1", "技能2"],
            "potential_challenges": ["挑戰1", "挑戰2"],
            "success_criteria": ["成功標準1", "成功標準2"],
            "recommended_approach": "建議處理方式"
        }}
        """
        
        try:
            response = self.pro_model.generate_content(prompt)
            return json.loads(response.text)
        except Exception as e:
            return self._fallback_analysis(user_input)
    
    def process_with_cli_integration(self, task_type: str, content: str) -> str:
        """整合 Gemini CLI 進行特定任務處理"""
        try:
            with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False) as temp_file:
                temp_file.write(content)
                temp_path = temp_file.name
            
            # 根據任務類型選擇 CLI 指令
            cli_commands = {
                "email_composition": f"gemini ask --creative 'Based on this content, write a professional email: {content}'",
                "meeting_summary": f"gemini analyze {temp_path} 'Summarize this meeting content with key points and action items'",
                "data_analysis": f"gemini ask --detailed 'Analyze this data and provide insights: {content}'",
                "creative_writing": f"gemini ask --creative 'Create engaging content based on: {content}'"
            }
            
            command = cli_commands.get(task_type, f"gemini ask '{content}'")
            result = subprocess.run(command, shell=True, capture_output=True, text=True)
            
            os.unlink(temp_path)  # 清理暫存檔
            
            return result.stdout if result.returncode == 0 else self._fallback_processing(content)
            
        except Exception as e:
            return self._fallback_processing(content)
    
    def multi_turn_refinement(self, initial_response: str, feedback: str) -> str:
        """多輪對話優化"""
        chat = self.pro_model.start_chat()
        
        # 第一輪：提供初始回應
        chat.send_message(f"我提供了以下回應：{initial_response}")
        
        # 第二輪：根據反饋改進
        refinement_prompt = f"""
        使用者對上述回應有以下反饋：{feedback}
        
        請根據反饋改進回應，讓它：
        1. 更符合使用者需求
        2. 更加專業和實用
        3. 提供更具體的建議
        """
        
        refined_response = chat.send_message(refinement_prompt)
        return refined_response.text
    
    def _fallback_analysis(self, user_input: str) -> Dict[str, Any]:
        """備用分析方法"""
        return {
            "task_type": "research_assistance",
            "priority_level": "medium",
            "key_requirements": [user_input],
            "estimated_duration": "15-30分鐘",
            "required_skills": ["資料分析"],
            "potential_challenges": ["需求不明確"],
            "success_criteria": ["提供有用資訊"],
            "recommended_approach": "逐步分析和回應"
        }
    
    def _fallback_processing(self, content: str) -> str:
        """備用處理方法"""
        try:
            response = self.pro_model.generate_content(f"請協助處理：{content}")
            return response.text
        except:
            return "抱歉，處理過程中發生錯誤，請稍後再試。"

3. 綜合工作流程 (workflows/email_workflow.py)

from langgraph.graph import StateGraph, END
from core.state_manager import WorkflowState
from integrations.gemini_integration import AdvancedGeminiClient
from typing import Literal

def analyze_email_request(state: WorkflowState) -> WorkflowState:
    """分析郵件撰寫需求"""
    client = AdvancedGeminiClient()
    
    analysis = client.analyze_work_request(state)
    
    return {
        **state,
        "gemini_analysis": analysis,
        "current_node": "email_analysis",
        "processing_history": state["processing_history"] + ["email_analysis"]
    }

def draft_email_content(state: WorkflowState) -> WorkflowState:
    """撰寫郵件內容"""
    client = AdvancedGeminiClient()
    
    # 構建詳細的郵件撰寫提示
    email_prompt = f"""
    根據以下資訊撰寫專業郵件：
    
    需求：{state['user_input']}
    分析結果：{state['gemini_analysis']}
    使用者偏好：{state['user_preferences']}
    
    請撰寫包含以下元素的郵件：
    1. 適當的主旨行
    2. 專業的開頭問候
    3. 清晰的內容主體
    4. 合適的結尾和署名
    
    郵件風格應該：{state['gemini_analysis'].get('recommended_approach', '專業簡潔')}
    """
    
    # 使用 CLI 整合處理
    draft = client.process_with_cli_integration("email_composition", email_prompt)
    
    return {
        **state,
        "intermediate_results": state["intermediate_results"] + [{"step": "draft", "content": draft}],
        "current_node": "email_draft",
        "processing_history": state["processing_history"] + ["email_draft"]
    }

def review_and_refine(state: WorkflowState) -> WorkflowState:
    """檢視和優化郵件"""
    client = AdvancedGeminiClient()
    
    draft = state["intermediate_results"][-1]["content"]
    
    review_prompt = f"""
    請檢視以下郵件草稿並提供改進建議：
    
    草稿：{draft}
    
    檢視重點：
    1. 語氣是否適當
    2. 內容是否完整
    3. 格式是否規範
    4. 是否有錯字或語法問題
    
    請提供改進後的版本。
    """
    
    refined_email = client.multi_turn_refinement(draft, review_prompt)
    
    return {
        **state,
        "final_output": refined_email,
        "confidence_score": 0.9,
        "suggestions": ["檢查收件者資訊", "確認附件", "預定發送時間"],
        "next_actions": ["傳送郵件", "儲存草稿", "進一步修改"],
        "current_node": "email_complete",
        "processing_history": state["processing_history"] + ["email_review"]
    }

def route_email_processing(state: WorkflowState) -> Literal["draft", "review", "complete"]:
    """路由郵件處理步驟"""
    current_node = state["current_node"]
    
    if current_node == "email_analysis":
        return "draft"
    elif current_node == "email_draft":
        return "review"
    else:
        return "complete"

def create_email_workflow():
    """建立郵件處理工作流程"""
    workflow = StateGraph(WorkflowState)
    
    # 添加節點
    workflow.add_node("analyze", analyze_email_request)
    workflow.add_node("draft", draft_email_content)
    workflow.add_node("review", review_and_refine)
    
    # 設定流程
    workflow.set_entry_point("analyze")
    
    workflow.add_conditional_edges(
        "analyze",
        route_email_processing,
        {"draft": "draft"}
    )
    
    workflow.add_conditional_edges(
        "draft", 
        route_email_processing,
        {"review": "review"}
    )
    
    workflow.add_edge("review", END)
    
    return workflow.compile()

4. 主程式整合 (main.py)

from core.state_manager import StateManager, WorkflowState
from workflows.email_workflow import create_email_workflow
from integrations.gemini_integration import AdvancedGeminiClient
import json

class SmartWorkAssistant:
    def __init__(self):
        self.state_manager = StateManager()
        self.gemini_client = AdvancedGeminiClient()
        self.workflows = {
            "email_composition": create_email_workflow(),
            # 其他工作流程可以在這裡添加
        }
    
    def process_request(self, user_input: str, attachments: list = None) -> Dict:
        """處理工作請求"""
        # 初始化狀態
        initial_state = {
            **self.state_manager.current_state,
            "user_input": user_input,
            "attachments": attachments or []
        }
        
        # 分析請求類型
        analysis = self.gemini_client.analyze_work_request(initial_state)
        task_type = analysis.get("task_type", "research_assistance")
        
        # 更新狀態
        self.state_manager.update_state({
            "task_type": task_type,
            "gemini_analysis": analysis
        })
        
        # 選擇對應的工作流程
        if task_type in self.workflows:
            workflow = self.workflows[task_type]
            result = workflow.invoke(self.state_manager.current_state)
        else:
            # 使用通用處理
            result = self._generic_processing(user_input)
        
        return {
            "response": result.get("final_output", "處理完成"),
            "task_type": task_type,
            "confidence": result.get("confidence_score", 0.8),
            "suggestions": result.get("suggestions", []),
            "next_actions": result.get("next_actions", [])
        }
    
    def _generic_processing(self, user_input: str) -> Dict:
        """通用處理方法"""
        response = self.gemini_client.process_with_cli_integration(
            "research_assistance", 
            user_input
        )
        
        return {
            "final_output": response,
            "confidence_score": 0.7,
            "suggestions": ["考慮提供更多細節", "嘗試不同的表達方式"],
            "next_actions": ["繼續對話", "尋求進一步協助"]
        }

def main():
    """主程式"""
    print("🚀 智能工作助理已啟動！")
    print("💼 我可以協助您：撰寫郵件、總結會議、分析資料、創意寫作等")
    print("=" * 70)
    
    assistant = SmartWorkAssistant()
    
    while True:
        try:
            user_input = input("\n📝 請描述您需要的協助：").strip()
            
            if not user_input:
                continue
            
            if user_input.lower() in ['quit', 'exit', '退出']:
                print("👋 再見！祝您工作順利！")
                break
            
            print("\n🤖 正在處理您的請求...")
            
            # 處理請求
            result = assistant.process_request(user_input)
            
            # 顯示結果
            print(f"\n📋 任務類型：{result['task_type']}")
            print(f"🎯 信心度：{result['confidence']:.2f}")
            print(f"\n✨ 處理結果：\n{result['response']}")
            
            if result['suggestions']:
                print(f"\n💡 建議：")
                for suggestion in result['suggestions']:
                    print(f"  • {suggestion}")
            
            if result['next_actions']:
                print(f"\n🎯 建議下一步：")
                for action in result['next_actions']:
                    print(f"  • {action}")
            
            print("\n" + "="*70)
            
        except KeyboardInterrupt:
            print("\n👋 再見！祝您工作順利！")
            break
        except Exception as e:
            print(f"\n❌ 發生錯誤：{e}")
            print("請再試一次。")

if __name__ == "__main__":
    main()

🎯 本週成就總結

技術掌握度

• ✅ Gemini API：從基礎呼叫到複雜整合 (90%)
• ✅ Gemini CLI：命令列操作和自動化 (85%)
• ✅ LangGraph：圖形化工作流程設計 (80%)
• ✅ 系統整合：多技術棧協同工作 (75%)

實戰專案

• 🤖 簡易聊天機器人：具備記憶和指令系統
• 🎓 智能學習助手：多節點工作流程應用
• 💼 綜合工作助理：深度整合 Gemini 和 LangGraph

🚀 下週預習

下週我們將進入第二階段的學習：

• 多輪對話與上下文管理
• 記憶功能和個性化
• 外部 API 整合
• 文件處理和知識庫

💡 學習心得與建議

1. 循序漸進：從簡單到複雜，每天的學習都是累積
2. 實戰為主：理論結合實際專案，更容易理解和記憶
3. 模組化思維：將複雜功能拆解成小模組，便於開發和除錯
4. 持續實驗：多嘗試不同的參數和方法，找到最適合的解決方案

🎪 週末作業

1. 優化現有專案：為你的聊天機器人添加更多功能
2. 探索新應用：思考 LangGraph 在你的工作中可以解決什麼問題
3. 技術實驗：嘗試不同的 Gemini 參數設定，觀察效果差異
4. 準備下週：思考你想要建立什麼樣的 AI 助理功能

恭喜完成第一週的學習！你已經具備了 AI 助理開發的基礎技能，下週我們將進入更進階的功能開發。加油！💪