今天我們要讓 AI 助理變得更加實用！透過整合外部 API 服務，我們的助理將能夠獲取即時資訊、執行實際任務，從一個單純的對話機器人進化為真正的數位助手。
AI 模型雖然強大，但它們的知識是固定的，無法獲取：
透過 API 整合，我們的助理將擁有「手和眼」，能夠真正幫助使用者完成實際任務。
api_integration_assistant/
├── main.py # 主程式入口
├── core/
│ ├── __init__.py
│ ├── api_manager.py # API 管理核心
│ └── integration_state.py # 整合狀態定義
├── services/
│ ├── __init__.py
│ ├── weather_service.py # 天氣服務
│ ├── news_service.py # 新聞服務
│ ├── translation_service.py # 翻譯服務
│ └── calculation_service.py # 計算服務
├── workflows/
│ ├── __init__.py
│ └── api_workflow.py # LangGraph API 工作流程
├── utils/
│ ├── __init__.py
│ ├── http_client.py # HTTP 客戶端
│ └── response_formatter.py # 回應格式化
└── config/
├── __init__.py
└── api_config.py # API 設定檔
import os
from typing import Dict, Any
from dotenv import load_dotenv
load_dotenv()
# API 金鑰設定
API_KEYS = {
'openweather': os.getenv('OPENWEATHER_API_KEY'),
'news': os.getenv('NEWS_API_KEY'),
'google_translate': os.getenv('GOOGLE_TRANSLATE_API_KEY'),
'gemini': os.getenv('GEMINI_API_KEY')
}
# API 端點設定
API_ENDPOINTS = {
'weather': {
'base_url': 'https://api.openweathermap.org/data/2.5',
'timeout': 10,
'retry_count': 3
},
'news': {
'base_url': 'https://newsapi.org/v2',
'timeout': 15,
'retry_count': 2
},
'translation': {
'base_url': 'https://translation.googleapis.com/language/translate/v2',
'timeout': 8,
'retry_count': 2
}
}
# 服務能力映射
SERVICE_CAPABILITIES = {
'weather': ['天氣', '氣溫', '降雨', '風速', '濕度', 'weather', 'temperature'],
'news': ['新聞', '資訊', '消息', '時事', 'news', 'information'],
'translation': ['翻譯', '轉換', '語言', 'translate', 'language'],
'calculation': ['計算', '數學', '算式', 'calculate', 'math']
}
import requests
import time
from typing import Dict, Any, Optional
import json
class RobustHTTPClient:
"""強健的 HTTP 客戶端，具備重試和錯誤處理機制"""
def __init__(self, timeout: int = 10, retry_count: int = 3):
self.timeout = timeout
self.retry_count = retry_count
self.session = requests.Session()
def get(self, url: str, params: Dict = None, headers: Dict = None) -> Optional[Dict]:
"""GET 請求"""
return self._make_request('GET', url, params=params, headers=headers)
def post(self, url: str, data: Dict = None, headers: Dict = None) -> Optional[Dict]:
"""POST 請求"""
return self._make_request('POST', url, json=data, headers=headers)
def _make_request(self, method: str, url: str, **kwargs) -> Optional[Dict]:
"""執行 HTTP 請求，具備重試機制"""
last_exception = None
for attempt in range(self.retry_count):
try:
response = self.session.request(
method=method,
url=url,
timeout=self.timeout,
**kwargs
)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
last_exception = e
if attempt < self.retry_count - 1:
# 指數退避重試
wait_time = (2 ** attempt) + 1
print(f"請求失敗，{wait_time}秒後重試... (嘗試 {attempt + 1}/{self.retry_count})")
time.sleep(wait_time)
continue
except json.JSONDecodeError as e:
print(f"JSON 解析錯誤: {e}")
return None
print(f"所有重試均失敗: {last_exception}")
return None
from utils.http_client import RobustHTTPClient
from config.api_config import API_KEYS, API_ENDPOINTS
from typing import Dict, Optional
class WeatherService:
"""天氣服務整合"""
def __init__(self):
self.client = RobustHTTPClient(**API_ENDPOINTS['weather'])
self.api_key = API_KEYS['openweather']
self.base_url = API_ENDPOINTS['weather']['base_url']
def get_current_weather(self, city: str, country: str = None) -> Optional[Dict]:
"""獲取當前天氣資訊"""
if not self.api_key:
return {"error": "OpenWeather API key not configured"}
# 構建查詢參數
location = f"{city},{country}" if country else city
params = {
'q': location,
'appid': self.api_key,
'units': 'metric', # 攝氏度
'lang': 'zh_tw' # 繁體中文
}
url = f"{self.base_url}/weather"
result = self.client.get(url, params=params)
if result and 'weather' in result:
return self._format_weather_response(result)
else:
return {"error": f"無法獲取 {city} 的天氣資訊"}
def get_weather_forecast(self, city: str, days: int = 5) -> Optional[Dict]:
"""獲取天氣預報"""
if not self.api_key:
return {"error": "OpenWeather API key not configured"}
params = {
'q': city,
'appid': self.api_key,
'units': 'metric',
'lang': 'zh_tw',
'cnt': min(days * 8, 40) # 每天8個時段，最多5天
}
url = f"{self.base_url}/forecast"
result = self.client.get(url, params=params)
if result and 'list' in result:
return self._format_forecast_response(result, days)
else:
return {"error": f"無法獲取 {city} 的天氣預報"}
def _format_weather_response(self, data: Dict) -> Dict:
"""格式化天氣回應"""
weather = data['weather'][0]
main = data['main']
wind = data.get('wind', {})
return {
"city": data['name'],
"country": data['sys']['country'],
"description": weather['description'],
"temperature": round(main['temp']),
"feels_like": round(main['feels_like']),
"humidity": main['humidity'],
"pressure": main['pressure'],
"wind_speed": wind.get('speed', 0),
"visibility": data.get('visibility', 0) // 1000, # 轉為公里
"formatted_response": self._create_weather_text(data)
}
def _format_forecast_response(self, data: Dict, days: int) -> Dict:
"""格式化預報回應"""
forecasts = []
processed_days = set()
for item in data['list']:
date_str = item['dt_txt'].split(' ')[0]
if date_str not in processed_days and len(processed_days) < days:
processed_days.add(date_str)
weather = item['weather'][0]
main = item['main']
forecasts.append({
"date": date_str,
"description": weather['description'],
"temp_max": round(main['temp_max']),
"temp_min": round(main['temp_min']),
"humidity": main['humidity']
})
return {
"city": data['city']['name'],
"forecasts": forecasts,
"formatted_response": self._create_forecast_text(data['city']['name'], forecasts)
}
def _create_weather_text(self, data: Dict) -> str:
"""創建天氣文字描述"""
weather = data['weather'][0]
main = data['main']
text = f"🌤️ **{data['name']} 當前天氣**\n\n"
text += f"🌡️ 溫度：{round(main['temp'])}°C (體感 {round(main['feels_like'])}°C)\n"
text += f"☁️ 天氣：{weather['description']}\n"
text += f"💧 濕度：{main['humidity']}%\n"
text += f"🌪️ 氣壓：{main['pressure']} hPa\n"
if 'wind' in data:
text += f"💨 風速：{data['wind'].get('speed', 0)} m/s\n"
return text
def _create_forecast_text(self, city: str, forecasts: list) -> str:
"""創建預報文字描述"""
text = f"📅 **{city} 未來天氣預報**\n\n"
for forecast in forecasts:
text += f"📆 {forecast['date']}\n"
text += f" 🌡️ {forecast['temp_min']}°C ~ {forecast['temp_max']}°C\n"
text += f" ☁️ {forecast['description']}\n"
text += f" 💧 濕度 {forecast['humidity']}%\n\n"
return text
from utils.http_client import RobustHTTPClient
import json
class TranslationService:
"""翻譯服務整合（使用免費的 LibreTranslate API）"""
def __init__(self):
self.client = RobustHTTPClient()
self.base_url = "https://libretranslate.de" # 免費的翻譯服務
def translate_text(self, text: str, target_lang: str, source_lang: str = "auto") -> Dict:
"""翻譯文字"""
# 語言代碼映射
lang_map = {
'中文': 'zh', '英文': 'en', '日文': 'ja', '韓文': 'ko',
'法文': 'fr', '德文': 'de', '西班牙文': 'es', '義大利文': 'it',
'chinese': 'zh', 'english': 'en', 'japanese': 'ja', 'korean': 'ko'
}
target_code = lang_map.get(target_lang.lower(), target_lang.lower())
source_code = lang_map.get(source_lang.lower(), source_lang.lower()) if source_lang != "auto" else "auto"
# 準備請求資料
data = {
"q": text,
"source": source_code if source_code != "auto" else "auto",
"target": target_code,
"format": "text"
}
try:
result = self.client.post(f"{self.base_url}/translate", data=data)
if result and 'translatedText' in result:
return {
"original_text": text,
"translated_text": result['translatedText'],
"source_language": source_lang,
"target_language": target_lang,
"formatted_response": self._format_translation(text, result['translatedText'], source_lang, target_lang)
}
else:
return {"error": "翻譯服務暫時無法使用"}
except Exception as e:
return {"error": f"翻譯失敗：{str(e)}"}
def detect_language(self, text: str) -> Dict:
"""檢測語言"""
data = {"q": text}
try:
result = self.client.post(f"{self.base_url}/detect", data=data)
if result and isinstance(result, list) and len(result) > 0:
detected = result[0]
return {
"language": detected.get('language', 'unknown'),
"confidence": detected.get('confidence', 0.0)
}
else:
return {"error": "無法檢測語言"}
except Exception as e:
return {"error": f"語言檢測失敗：{str(e)}"}
def _format_translation(self, original: str, translated: str, source: str, target: str) -> str:
"""格式化翻譯結果"""
return f"""🌐 **翻譯結果**
📝 原文 ({source})：
{original}
✨ 譯文 ({target})：
{translated}"""
from langgraph.graph import StateGraph, END
from typing import TypedDict, List, Dict, Any, Literal
from services.weather_service import WeatherService
from services.translation_service import TranslationService
import google.generativeai as genai
import os
import re
genai.configure(api_key=os.getenv('GEMINI_API_KEY'))
model = genai.GenerativeModel('gemini-2.5-flash')
class APIIntegrationState(TypedDict):
user_input: str
detected_services: List[str]
api_responses: Dict[str, Any]
final_response: str
processing_mode: str
confidence_score: float
# 初始化服務
weather_service = WeatherService()
translation_service = TranslationService()
def analyze_service_needs(state: APIIntegrationState) -> APIIntegrationState:
"""分析需要使用的服務"""
user_input = state["user_input"].lower()
detected_services = []
# 天氣服務檢測
weather_keywords = ['天氣', '氣溫', '下雨', '晴天', '陰天', 'weather', 'temperature']
if any(keyword in user_input for keyword in weather_keywords):
detected_services.append('weather')
# 翻譯服務檢測
translation_keywords = ['翻譯', '轉換', '英文', '中文', '日文', 'translate']
if any(keyword in user_input for keyword in translation_keywords):
detected_services.append('translation')
# 如果沒有檢測到特定服務，使用通用處理
if not detected_services:
detected_services.append('general')
return {
**state,
"detected_services": detected_services,
"processing_mode": "api_integration" if len(detected_services) > 1 else detected_services[0]
}
def process_weather_request(state: APIIntegrationState) -> APIIntegrationState:
"""處理天氣請求"""
user_input = state["user_input"]
# 提取城市名稱
city = extract_city_name(user_input)
if not city:
city = "台北" # 預設城市
# 檢查是否要預報
if '預報' in user_input or 'forecast' in user_input.lower():
weather_data = weather_service.get_weather_forecast(city)
else:
weather_data = weather_service.get_current_weather(city)
api_responses = state.get("api_responses", {})
api_responses["weather"] = weather_data
if weather_data and "formatted_response" in weather_data:
response = weather_data["formatted_response"]
confidence = 0.9
else:
response = f"抱歉，無法獲取 {city} 的天氣資訊。"
confidence = 0.3
return {
**state,
"api_responses": api_responses,
"final_response": response,
"confidence_score": confidence
}
def process_translation_request(state: APIIntegrationState) -> APIIntegrationState:
"""處理翻譯請求"""
user_input = state["user_input"]
# 提取要翻譯的文字和目標語言
translation_info = extract_translation_info(user_input)
if translation_info["text"] and translation_info["target_lang"]:
result = translation_service.translate_text(
translation_info["text"],
translation_info["target_lang"],
translation_info["source_lang"]
)
else:
result = {"error": "無法識別翻譯需求，請明確指定要翻譯的文字和目標語言"}
api_responses = state.get("api_responses", {})
api_responses["translation"] = result
if result and "formatted_response" in result:
response = result["formatted_response"]
confidence = 0.8
else:
response = result.get("error", "翻譯處理失敗")
confidence = 0.3
return {
**state,
"api_responses": api_responses,
"final_response": response,
"confidence_score": confidence
}
def process_general_request(state: APIIntegrationState) -> APIIntegrationState:
"""處理一般請求"""
user_input = state["user_input"]
try:
response = model.generate_content(user_input)
final_response = response.text
confidence = 0.7
except:
final_response = "抱歉，我暫時無法處理您的請求。請稍後再試。"
confidence = 0.3
return {
**state,
"final_response": final_response,
"confidence_score": confidence
}
def route_service_processing(state: APIIntegrationState) -> Literal["weather", "translation", "general"]:
"""路由服務處理"""
detected_services = state["detected_services"]
if "weather" in detected_services:
return "weather"
elif "translation" in detected_services:
return "translation"
else:
return "general"
def extract_city_name(text: str) -> str:
"""提取城市名稱"""
# 簡單的城市名稱提取
cities = ['台北', '台中', '台南', '高雄', '桃園', '新竹', '基隆', '台東', '花蓮', '嘉義', '彰化', '雲林', '南投', '宜蘭', '屏東', '澎湖', '金門', '連江']
for city in cities:
if city in text:
return city
# 使用正則表達式匹配可能的城市名稱
city_pattern = r'([a-zA-Z\u4e00-\u9fff]+(?:市|縣|區)?)'
matches = re.findall(city_pattern, text)
return matches[0] if matches else "台北"
def extract_translation_info(text: str) -> Dict[str, str]:
"""提取翻譯資訊"""
# 尋找翻譯模式
patterns = [
r'翻譯[「『"](.*?)[」』"](?:成|為|到)(\w+)',
r'把[「『"](.*?)[」』"]翻譯成(\w+)',
r'[「『"](.*?)[」』"](?:的)?(\w+)翻譯',
r'翻譯.*?[：:]\s*(.+?)(?:\s+(?:成|為|到)\s*(\w+))?'
]
for pattern in patterns:
match = re.search(pattern, text)
if match:
groups = match.groups()
return {
"text": groups[0].strip(),
"target_lang": groups[1] if len(groups) > 1 and groups[1] else "英文",
"source_lang": "auto"
}
# 如果沒有匹配到特定模式，嘗試簡單提取
if '翻譯' in text:
parts = text.split('翻譯')
if len(parts) > 1:
return {
"text": parts[0].strip(),
"target_lang": "英文",
"source_lang": "auto"
}
return {"text": "", "target_lang": "", "source_lang": "auto"}
def create_api_integration_workflow():
"""建立 API 整合工作流程"""
workflow = StateGraph(APIIntegrationState)
# 添加節點
workflow.add_node("analyze", analyze_service_needs)
workflow.add_node("weather", process_weather_request)
workflow.add_node("translation", process_translation_request)
workflow.add_node("general", process_general_request)
# 設定流程
workflow.set_entry_point("analyze")
# 條件路由
workflow.add_conditional_edges(
"analyze",
route_service_processing,
{
"weather": "weather",
"translation": "translation",
"general": "general"
}
)
# 結束節點
workflow.add_edge("weather", END)
workflow.add_edge("translation", END)
workflow.add_edge("general", END)
return workflow.compile()
from workflows.api_workflow import create_api_integration_workflow
from dotenv import load_dotenv
load_dotenv()
def main():
"""API 整合助理主程式"""
print("🌐 多功能 API 整合助理")
print("🔧 支援服務：天氣查詢、文字翻譯、一般問答")
print("💡 範例指令：")
print(" • '台北今天天氣如何？'")
print(" • '翻譯「Hello World」成中文'")
print(" • '未來三天台中的天氣預報'")
print("=" * 60)
# 建立工作流程
app = create_api_integration_workflow()
print("🤖 助理已就緒！輸入 'quit' 結束程式\n")
while True:
try:
user_input = input("💬 您：").strip()
if not user_input:
continue
if user_input.lower() in ['quit', 'exit', '退出']:
print("👋 再見！感謝使用 API 整合助理！")
break
print("🔍 分析請求中...")
# 執行 API 整合工作流程
initial_state = {
"user_input": user_input,
"detected_services": [],
"api_responses": {},
"final_response": "",
"processing_mode": "",
"confidence_score": 0.0
}
result = app.invoke(initial_state)
print(f"🤖 助理：{result['final_response']}")
# 顯示處理資訊
if result.get('detected_services'):
print(f"🔧 使用服務：{', '.join(result['detected_services'])}")
if result.get('confidence_score', 0) > 0:
print(f"🎯 信心度：{result['confidence_score']:.2f}")
print("-" * 50)
except KeyboardInterrupt:
print("\n👋 再見！感謝使用 API 整合助理！")
break
except Exception as e:
print(f"❌ 發生錯誤：{e}")
continue
if __name__ == "__main__":
main()
💬 您：台北今天天氣如何？
🔍 分析請求中...
🤖 助理：🌤️ **台北 當前天氣**
🌡️ 溫度：23°C (體感 25°C)
☁️ 天氣：多雲
💧 濕度：65%
🌪️ 氣壓：1013 hPa
💨 風速：3.2 m/s
💬 您：翻譯「Good morning」成中文
🔍 分析請求中...
🤖 助理：🌐 **翻譯結果**
📝 原文 (auto)：
Good morning
✨ 譯文 (中文)：
早安
✅ 多服務整合：天氣、翻譯、通用問答
✅ 智能路由：自動識別服務需求
✅ 強健客戶端：重試機制和錯誤處理
✅ 格式化回應：美觀的結果展示
✅ LangGraph 管理：清晰的工作流程控制
今天我們成功建立了一個多功能的 API 整合系統！AI 助理現在能夠：連接外部服務獲取即時資訊、提供實用的功能服務、智能路由到對應的處理模組。
明天我們將學習文件處理與知識庫建構，讓助理能夠理解和分析各種文件格式！
iThome鐵人賽
看影片追技術
看更多