feat: 完善词典子图,添加API调用和前端格式化工具
Some checks failed
构建并部署 AI Agent 服务 / deploy (push) Failing after 6m5s

- 完善词典子图:添加生词本功能
- 创建API调用工具:dictionary_api
- 添加前端格式化展示工具:result_formatter.py
- 创建通讯录和资讯子图的基本结构
- 更新主图状态结构,添加MainGraphState
- 添加subgraph_builder.py用于子图集成
This commit is contained in:
2026-04-25 18:29:23 +08:00
parent 03ba825645
commit a14744f18b
17 changed files with 2057 additions and 18 deletions

View File

@@ -0,0 +1,402 @@
"""
词典子图节点 - 完善版使用API客户端
Dictionary Subgraph Nodes - Complete (with API Client)
"""
from typing import Dict, Any, List
from datetime import datetime
import random
from .state import (
DictionaryState,
DictionaryAction,
WordEntry,
ExtractedTerm
)
from .api_client import dictionary_api
# ========== 模拟生词本存储(后续可替换为数据库) ==========
WORD_BOOK_DB: Dict[str, List[Dict]] = {} # user_id -> [word_entries]
def parse_intent(state: DictionaryState) -> DictionaryState:
"""
解析用户意图节点
确定用户想做什么操作
"""
state.current_phase = "intent_parsing"
query_lower = state.user_query.lower()
# 简单的关键词匹配
if any(keyword in query_lower for keyword in ["翻译", "translate", "英语", "英文"]):
state.action = DictionaryAction.TRANSLATE
state.action_params = {"text": state.user_query}
# 同时设置source_text
text = state.user_query
for keyword in ["翻译", "translate", "英语", "英文"]:
text = text.replace(keyword, "")
state.source_text = text.strip()
elif any(keyword in query_lower for keyword in ["查询", "query", "单词", "word"]):
state.action = DictionaryAction.QUERY
state.action_params = {"word": state.user_query}
elif any(keyword in query_lower for keyword in ["每日", "daily", "一词"]):
state.action = DictionaryAction.DAILY_WORD
elif any(keyword in query_lower for keyword in ["提取", "extract", "术语", "term"]):
state.action = DictionaryAction.EXTRACT
state.action_params = {"text": state.user_query}
elif any(keyword in query_lower for keyword in ["生词本", "wordbook", "我的单词"]):
state.action = DictionaryAction.WORD_BOOK_LOOKUP
elif any(keyword in query_lower for keyword in ["添加到生词本", "添加单词", "add to wordbook"]):
state.action = DictionaryAction.WORD_BOOK_ADD
state.action_params = {"word": state.user_query}
else:
# 默认翻译
state.action = DictionaryAction.TRANSLATE
state.source_text = state.user_query
return state
def query_word(state: DictionaryState) -> DictionaryState:
"""
查询单词节点
"""
state.current_phase = "querying_word"
word = state.action_params.get("word", state.user_query)
word = word.replace("查询", "").replace("单词", "").strip()
# 使用API客户端查询单词
data = dictionary_api.query_word_mock(word)
state.word_entry = WordEntry(
word=word,
phonetic=data.get("phonetic", ""),
part_of_speech=data.get("part_of_speech", "n."),
definitions=data.get("definitions", []),
examples=data.get("examples", []),
)
state.success = True
return state
def translate_text(state: DictionaryState) -> DictionaryState:
"""
翻译文本节点
"""
state.current_phase = "translating"
text = state.source_text or state.action_params.get("text", state.user_query)
# 使用API客户端翻译
data = dictionary_api.translate_mock(text, state.source_language, state.target_language)
state.translated_text = data.get("translated_text", f"【翻译结果】{text}")
state.translation_confidence = data.get("confidence", 0.95)
state.success = True
return state
def extract_terms(state: DictionaryState) -> DictionaryState:
"""
提取专业术语节点
"""
state.current_phase = "extracting_terms"
text = state.source_text or state.action_params.get("text", state.user_query)
# 使用API客户端提取术语
terms_data = dictionary_api.extract_terms_mock(text)
for term_data in terms_data:
state.extracted_terms.append(ExtractedTerm(
term=term_data.get("term", ""),
type=term_data.get("type", ""),
definition=term_data.get("definition", ""),
confidence=term_data.get("confidence", 0.0)
))
state.success = True
return state
def get_daily_word(state: DictionaryState) -> DictionaryState:
"""
获取每日一词节点
"""
state.current_phase = "getting_daily_word"
# 每日一词候选
words = ["serendipity", "ephemeral", "ubiquitous", "eloquent", "resilient"]
# 基于日期选择固定词,这样同一天的每日一词是固定的
day_of_year = datetime.now().timetuple().tm_yday
chosen_idx = day_of_year % len(words)
chosen_word = words[chosen_idx]
# 使用API客户端查询单词详情
data = dictionary_api.query_word_mock(chosen_word)
state.daily_word = WordEntry(
word=chosen_word,
phonetic=data.get("phonetic", ""),
part_of_speech=data.get("part_of_speech", "adj."),
definitions=data.get("definitions", []),
examples=data.get("examples", []),
)
state.success = True
return state
def lookup_word_book(state: DictionaryState) -> DictionaryState:
"""
查询生词本节点
"""
state.current_phase = "looking_up_wordbook"
user_id = state.user_id or "default_user"
word_book = WORD_BOOK_DB.get(user_id, [])
# 构建WordEntry列表
if word_book:
for entry in word_book:
state.extracted_terms.append(ExtractedTerm(
term=entry.get("word", ""),
type="生词本单词",
definition=entry.get("definitions", [""])[0] if entry.get("definitions") else "",
confidence=1.0
))
state.success = True
return state
def add_to_word_book(state: DictionaryState) -> DictionaryState:
"""
添加到生词本节点
"""
state.current_phase = "adding_to_wordbook"
user_id = state.user_id or "default_user"
word = state.action_params.get("word", state.user_query)
word = word.replace("添加到生词本", "").replace("添加单词", "").strip()
# 查询单词信息
query_state = DictionaryState(user_query=word, action=DictionaryAction.QUERY)
query_state = query_word(query_state)
if query_state.word_entry:
we = query_state.word_entry
# 添加到模拟数据库
if user_id not in WORD_BOOK_DB:
WORD_BOOK_DB[user_id] = []
# 检查是否已存在
exists = any(entry.get("word") == we.word for entry in WORD_BOOK_DB[user_id])
if not exists:
entry_dict = {
"word": we.word,
"phonetic": we.phonetic,
"part_of_speech": we.part_of_speech,
"definitions": we.definitions,
"examples": we.examples,
"added_at": datetime.now().isoformat(),
"review_count": 0,
"next_review_at": None
}
WORD_BOOK_DB[user_id].append(entry_dict)
state.final_result = f"✅ 已将 '{we.word}' 添加到生词本!"
else:
state.final_result = f" '{we.word}' 已在生词本中!"
state.success = True
return state
def format_result(state: DictionaryState) -> DictionaryState:
"""
格式化结果节点 - 精美输出
"""
state.current_phase = "formatting"
if state.action == DictionaryAction.QUERY and state.word_entry:
we = state.word_entry
result = []
result.append("═══════════════════════════════════════════")
result.append("📚 单词查询结果")
result.append("═══════════════════════════════════════════")
result.append(f"")
result.append(f" {we.word}")
if we.phonetic:
result.append(f" {we.phonetic}")
result.append(f"{we.part_of_speech}")
result.append(f"")
result.append("📖 释义:")
for i, definition in enumerate(we.definitions, 1):
result.append(f" {i}. {definition}")
if we.examples:
result.append("")
result.append("💡 例句:")
for example in we.examples:
result.append(f" \"{example}\"")
if we.synonyms:
result.append("")
result.append("🔗 同义词:")
result.append(f" {', '.join(we.synonyms)}")
if we.antonyms:
result.append("")
result.append("🔗 反义词:")
result.append(f" {', '.join(we.antonyms)}")
result.append("")
result.append("═══════════════════════════════════════════")
result.append("💡 提示:回复 '添加到生词本 + 单词' 可收藏")
state.final_result = "\n".join(result)
elif state.action == DictionaryAction.TRANSLATE:
result = []
result.append("═══════════════════════════════════════════")
result.append("🔄 翻译结果")
result.append("═══════════════════════════════════════════")
result.append(f"")
result.append(f" 原文:{state.source_text}")
result.append(f" 译文:{state.translated_text}")
result.append(f"")
result.append(f" 🎯 置信度:{state.translation_confidence:.0%}")
result.append("")
result.append("═══════════════════════════════════════════")
state.final_result = "\n".join(result)
elif state.action == DictionaryAction.DAILY_WORD and state.daily_word:
dw = state.daily_word
result = []
result.append("═══════════════════════════════════════════")
result.append("🌟 每日一词")
result.append("═══════════════════════════════════════════")
result.append(f"")
result.append(f" {dw.word}")
if dw.phonetic:
result.append(f" {dw.phonetic}")
result.append(f"{dw.part_of_speech}")
result.append(f"")
if dw.definitions:
result.append("📖 释义:")
for i, definition in enumerate(dw.definitions, 1):
result.append(f" {i}. {definition}")
if dw.examples:
result.append("")
result.append("💡 例句:")
for example in dw.examples:
result.append(f" \"{example}\"")
result.append("")
result.append("═══════════════════════════════════════════")
result.append("💡 学习提示:尝试用这个词造一个句子")
result.append("💡 收藏提示:回复 '添加到生词本' 可收藏")
state.final_result = "\n".join(result)
elif state.action == DictionaryAction.EXTRACT and state.extracted_terms:
result = []
result.append("═══════════════════════════════════════════")
result.append("📋 提取的术语")
result.append("═══════════════════════════════════════════")
result.append("")
for i, term in enumerate(state.extracted_terms, 1):
result.append(f" {i}. {term.term}{term.type}")
result.append(f" {term.definition}")
result.append(f" 🎯 置信度:{term.confidence:.0%}")
result.append("")
result.append("═══════════════════════════════════════════")
state.final_result = "\n".join(result)
elif state.action == DictionaryAction.WORD_BOOK_LOOKUP:
user_id = state.user_id or "default_user"
word_book = WORD_BOOK_DB.get(user_id, [])
result = []
result.append("═══════════════════════════════════════════")
result.append("📚 我的生词本")
result.append("═══════════════════════════════════════════")
result.append(f"")
if word_book:
result.append(f"{len(word_book)} 个单词")
result.append("")
for i, entry in enumerate(word_book, 1):
word = entry.get("word", "")
added_at = entry.get("added_at", "")
added_date = datetime.fromisoformat(added_at).strftime("%Y-%m-%d") if added_at else ""
result.append(f" {i}. {word} (添加于:{added_date}")
else:
result.append(" 生词本为空")
result.append(" 💡 提示:查询单词后可添加到生词本")
result.append("")
result.append("═══════════════════════════════════════════")
state.final_result = "\n".join(result)
elif state.action == DictionaryAction.WORD_BOOK_ADD:
if state.final_result:
result = state.final_result
else:
result = "添加完成"
state.final_result = result
else:
if not state.final_result:
state.final_result = "词典操作完成"
state.current_phase = "done"
return state
def should_continue(state: DictionaryState) -> str:
"""
条件路由:决定下一步该做什么
"""
if state.error_message:
return "format_result"
# 根据action路由
if state.action == DictionaryAction.NONE:
return "parse_intent"
elif state.action == DictionaryAction.QUERY:
return "query_word"
elif state.action == DictionaryAction.TRANSLATE:
return "translate_text"
elif state.action == DictionaryAction.EXTRACT:
return "extract_terms"
elif state.action == DictionaryAction.DAILY_WORD:
return "get_daily_word"
elif state.action == DictionaryAction.WORD_BOOK_LOOKUP:
return "lookup_word_book"
elif state.action == DictionaryAction.WORD_BOOK_ADD:
return "add_to_word_book"
else:
return "format_result"