root
9c53f58165
All checks were successful
构建并部署 AI Agent 服务 / deploy (push) Successful in 5m38s
- 添加MCP Manager统一入口管理 - 实现Contact/Dictionary/News三个适配器 - 三层降级策略：MCP -> Database -> Mock - 保持原有api_client向后兼容 - 添加完整文档和测试
140 lines
5.1 KiB
Python
140 lines
5.1 KiB
Python
"""
|
||
词典适配器
|
||
整合MCP、数据库缓存和模拟数据
|
||
"""
|
||
from typing import Dict, Any, Optional, List
|
||
from datetime import datetime
|
||
from .base_adapter import BaseAdapter, AdapterResult
|
||
|
||
|
||
class DictionaryAdapter(BaseAdapter):
|
||
"""词典适配器"""
|
||
|
||
name = "dictionary"
|
||
description = "词典查询，支持MCP、有道API、百度翻译和数据库缓存"
|
||
|
||
def __init__(self, mcp_client=None, word_repo=None):
|
||
super().__init__(mcp_client, word_repo)
|
||
self._mock_db = {
|
||
"serendipity": {
|
||
"phonetic": "/ˌserənˈdipədē/",
|
||
"part_of_speech": "n.",
|
||
"definitions": ["意外发现珍奇事物的能力", "机缘凑巧"],
|
||
"examples": ["Finding that old photo was pure serendipity."]
|
||
},
|
||
"ephemeral": {
|
||
"phonetic": "/əˈfem(ə)rəl/",
|
||
"part_of_speech": "adj.",
|
||
"definitions": ["短暂的，瞬息的"],
|
||
"examples": ["Fame in the digital age is often ephemeral."]
|
||
}
|
||
}
|
||
|
||
async def execute(self, action: str, **kwargs) -> AdapterResult:
|
||
"""统一执行入口"""
|
||
user_id = kwargs.get("user_id", "default")
|
||
word = kwargs.get("word", "")
|
||
use_cache = kwargs.get("use_cache", True)
|
||
|
||
# 1. 先查缓存
|
||
if use_cache and self.repository and word:
|
||
cached = await self._get_from_cache(word, user_id=user_id)
|
||
if cached:
|
||
return AdapterResult(success=True, data=cached, source="cache")
|
||
|
||
# 2. 尝试MCP
|
||
if self.mcp_client and self.mcp_client.is_available():
|
||
try:
|
||
mcp_result = await self._execute_mcp(action, **kwargs)
|
||
if mcp_result.success:
|
||
if use_cache and word:
|
||
await self._save_to_cache(word, mcp_result.data, user_id=user_id)
|
||
return mcp_result
|
||
except Exception as e:
|
||
print(f"[Dictionary] MCP调用失败: {e}")
|
||
|
||
# 3. 尝试第三方API（预留）
|
||
# result = await self._execute_api(action, **kwargs)
|
||
|
||
# 4. 降级到模拟数据
|
||
result = self._fallback(action, **kwargs)
|
||
if use_cache and word and result.success:
|
||
await self._save_to_cache(word, result.data, user_id=user_id)
|
||
return result
|
||
|
||
async def _execute_mcp(self, action: str, **kwargs) -> AdapterResult:
|
||
"""通过MCP执行"""
|
||
if action == "query_word":
|
||
word = kwargs.get("word", "")
|
||
result = await self.mcp_client.call_tool(
|
||
"dictionary_lookup_word",
|
||
{"word": word}
|
||
)
|
||
if result.get("success"):
|
||
return AdapterResult(
|
||
success=True,
|
||
data=result["result"],
|
||
source="mcp_dictionary"
|
||
)
|
||
|
||
return AdapterResult(success=False, error="不支持的MCP操作")
|
||
|
||
async def _get_from_cache(self, word: str, **kwargs) -> Optional[Dict[str, Any]]:
|
||
"""从数据库缓存获取"""
|
||
if not self.repository:
|
||
return None
|
||
try:
|
||
# 数据库查询（可选功能）
|
||
return None
|
||
except Exception as e:
|
||
print(f"[Dictionary] 缓存查询失败: {e}")
|
||
return None
|
||
|
||
async def _save_to_cache(self, word: str, data: Dict[str, Any], **kwargs):
|
||
"""保存到数据库缓存"""
|
||
if not self.repository:
|
||
return
|
||
try:
|
||
# 数据库保存（可选功能）
|
||
pass
|
||
except Exception as e:
|
||
print(f"[Dictionary] 缓存保存失败: {e}")
|
||
|
||
def _get_mock_data(self, action: str, **kwargs) -> Any:
|
||
"""获取模拟数据"""
|
||
if action == "query_word":
|
||
word = kwargs.get("word", "").lower()
|
||
if word in self._mock_db:
|
||
result = self._mock_db[word].copy()
|
||
result["word"] = word
|
||
return result
|
||
else:
|
||
return {
|
||
"word": word,
|
||
"phonetic": "",
|
||
"part_of_speech": "n.",
|
||
"definitions": [f"{word} 的释义1", f"{word} 的释义2"],
|
||
"examples": [f"This is an example sentence with '{word}'."]
|
||
}
|
||
|
||
elif action == "translate":
|
||
text = kwargs.get("text", "")
|
||
translations = {
|
||
"你好": "Hello",
|
||
"hello": "你好",
|
||
"人工智能": "Artificial Intelligence",
|
||
}
|
||
return {
|
||
"translated_text": translations.get(text.lower(), f"【翻译】{text}"),
|
||
"confidence": 0.95
|
||
}
|
||
|
||
elif action == "extract_terms":
|
||
text = kwargs.get("text", "")
|
||
return [
|
||
{"term": "AI", "type": "技术术语", "definition": "人工智能", "confidence": 0.95},
|
||
{"term": "大模型", "type": "技术术语", "definition": "大语言模型", "confidence": 0.92}
|
||
]
|
||
|
||
return None
|