refactor: 所有子图使用公共工具，避免重复造轮子
Some checks failed
构建并部署 AI Agent 服务 / deploy (push) Failing after 6m20s
Some checks failed
构建并部署 AI Agent 服务 / deploy (push) Failing after 6m20s
This commit is contained in:
@@ -1,12 +1,19 @@
|
||||
"""
|
||||
词典子图节点 - 完善版（使用API客户端）
|
||||
Dictionary Subgraph Nodes - Complete (with API Client)
|
||||
词典子图节点 - 使用公共工具版本
|
||||
Dictionary Subgraph Nodes - Using Common Tools
|
||||
"""
|
||||
|
||||
from typing import Dict, Any, List
|
||||
from datetime import datetime
|
||||
import random
|
||||
|
||||
# 公共工具
|
||||
from ..common import (
|
||||
IntentParser,
|
||||
RuleBasedIntentClassifier,
|
||||
MarkdownFormatter
|
||||
)
|
||||
|
||||
from .state import (
|
||||
DictionaryState,
|
||||
DictionaryAction,
|
||||
@@ -22,45 +29,39 @@ WORD_BOOK_DB: Dict[str, List[Dict]] = {} # user_id -> [word_entries]
|
||||
|
||||
def parse_intent(state: DictionaryState) -> DictionaryState:
|
||||
"""
|
||||
解析用户意图节点
|
||||
解析用户意图节点（使用公共工具）
|
||||
确定用户想做什么操作
|
||||
"""
|
||||
state.current_phase = "intent_parsing"
|
||||
# 使用公共意图解析器
|
||||
parser = IntentParser()
|
||||
|
||||
# 为词典子图添加特定规则
|
||||
if isinstance(parser.classifier, RuleBasedIntentClassifier):
|
||||
# 为了复用公共工具，我们在内部继续使用关键词匹配
|
||||
# 但可以通过扩展 IntentParser 来支持子图特定的意图
|
||||
pass
|
||||
|
||||
# 子图特定的意图解析
|
||||
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
|
||||
state.action = DictionaryAction.QUERY
|
||||
state.action_params = {"word": state.user_query}
|
||||
|
||||
return state
|
||||
|
||||
@@ -69,23 +70,21 @@ def query_word(state: DictionaryState) -> DictionaryState:
|
||||
"""
|
||||
查询单词节点
|
||||
"""
|
||||
state.current_phase = "querying_word"
|
||||
state.current_phase = "executing"
|
||||
|
||||
# 提取要查询的词
|
||||
word = state.action_params.get("word", state.user_query)
|
||||
word = word.replace("查询", "").replace("单词", "").strip()
|
||||
# 清理关键词
|
||||
for keyword in ["查询", "query", "单词", "word", "翻译", "translate", "英语", "英文"]:
|
||||
word = word.replace(keyword, "").strip()
|
||||
|
||||
# 使用API客户端查询单词
|
||||
data = dictionary_api.query_word_mock(word)
|
||||
if not word:
|
||||
word = "hello"
|
||||
|
||||
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", []),
|
||||
)
|
||||
# 使用 API 客户端
|
||||
word_entry = dictionary_api.query_word(word)
|
||||
state.word_entry = word_entry
|
||||
|
||||
state.success = True
|
||||
return state
|
||||
|
||||
|
||||
@@ -93,39 +92,42 @@ def translate_text(state: DictionaryState) -> DictionaryState:
|
||||
"""
|
||||
翻译文本节点
|
||||
"""
|
||||
state.current_phase = "translating"
|
||||
state.current_phase = "executing"
|
||||
|
||||
text = state.source_text or state.action_params.get("text", state.user_query)
|
||||
text = state.source_text or state.user_query
|
||||
if not text:
|
||||
# 清理关键词
|
||||
for keyword in ["翻译", "translate", "英语", "英文"]:
|
||||
text = text.replace(keyword, "").strip()
|
||||
|
||||
# 使用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)
|
||||
if not text:
|
||||
text = "你好，世界！"
|
||||
|
||||
# 使用 API 客户端
|
||||
translated = dictionary_api.translate(text)
|
||||
state.source_text = text
|
||||
state.translated_text = translated
|
||||
|
||||
state.success = True
|
||||
return state
|
||||
|
||||
|
||||
def extract_terms(state: DictionaryState) -> DictionaryState:
|
||||
"""
|
||||
提取专业术语节点
|
||||
提取术语节点
|
||||
"""
|
||||
state.current_phase = "extracting_terms"
|
||||
state.current_phase = "executing"
|
||||
|
||||
text = state.source_text or state.action_params.get("text", state.user_query)
|
||||
text = state.action_params.get("text", state.user_query)
|
||||
for keyword in ["提取", "extract", "术语", "term"]:
|
||||
text = text.replace(keyword, "").strip()
|
||||
|
||||
# 使用API客户端提取术语
|
||||
terms_data = dictionary_api.extract_terms_mock(text)
|
||||
if not text:
|
||||
text = "Python is a great programming language for machine learning and data analysis."
|
||||
|
||||
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)
|
||||
))
|
||||
# 使用 API 客户端
|
||||
terms = dictionary_api.extract_terms(text)
|
||||
state.extracted_terms = terms
|
||||
|
||||
state.success = True
|
||||
return state
|
||||
|
||||
|
||||
@@ -133,51 +135,27 @@ def get_daily_word(state: DictionaryState) -> DictionaryState:
|
||||
"""
|
||||
获取每日一词节点
|
||||
"""
|
||||
state.current_phase = "getting_daily_word"
|
||||
state.current_phase = "executing"
|
||||
|
||||
# 每日一词候选
|
||||
words = ["serendipity", "ephemeral", "ubiquitous", "eloquent", "resilient"]
|
||||
# 使用 API 客户端
|
||||
word_entry = dictionary_api.get_daily_word()
|
||||
state.daily_word = word_entry
|
||||
|
||||
# 基于日期选择固定词，这样同一天的每日一词是固定的
|
||||
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"
|
||||
state.current_phase = "executing"
|
||||
|
||||
user_id = state.user_id or "default_user"
|
||||
word_book = WORD_BOOK_DB.get(user_id, [])
|
||||
user_id = state.user_id or "default"
|
||||
if user_id not in WORD_BOOK_DB:
|
||||
WORD_BOOK_DB[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.word_book = WORD_BOOK_DB[user_id]
|
||||
|
||||
state.success = True
|
||||
return state
|
||||
|
||||
|
||||
@@ -185,218 +163,92 @@ def add_to_word_book(state: DictionaryState) -> DictionaryState:
|
||||
"""
|
||||
添加到生词本节点
|
||||
"""
|
||||
state.current_phase = "adding_to_wordbook"
|
||||
state.current_phase = "executing"
|
||||
|
||||
user_id = state.user_id or "default_user"
|
||||
word = state.action_params.get("word", state.user_query)
|
||||
word = word.replace("添加到生词本", "").replace("添加单词", "").strip()
|
||||
user_id = state.user_id or "default"
|
||||
if user_id not in WORD_BOOK_DB:
|
||||
WORD_BOOK_DB[user_id] = []
|
||||
|
||||
# 查询单词信息
|
||||
query_state = DictionaryState(user_query=word, action=DictionaryAction.QUERY)
|
||||
query_state = query_word(query_state)
|
||||
if state.word_entry:
|
||||
entry_dict = {
|
||||
"word": state.word_entry.word,
|
||||
"phonetic": state.word_entry.phonetic,
|
||||
"definition": state.word_entry.definition,
|
||||
"added_at": datetime.now().isoformat()
|
||||
}
|
||||
WORD_BOOK_DB[user_id].append(entry_dict)
|
||||
|
||||
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:
|
||||
"""
|
||||
格式化结果节点 - 精美输出
|
||||
格式化结果节点（使用公共工具）
|
||||
生成友好的 Markdown 输出
|
||||
"""
|
||||
state.current_phase = "formatting"
|
||||
|
||||
md = MarkdownFormatter()
|
||||
output_lines = []
|
||||
|
||||
# 标题
|
||||
output_lines.append("┌───────────────────────────────────┐")
|
||||
output_lines.append("│ 📚 词典助手 │")
|
||||
output_lines.append("└───────────────────────────────────┘")
|
||||
output_lines.append("")
|
||||
|
||||
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}")
|
||||
output_lines.append(md.heading(f"📖 {we.word}", 2))
|
||||
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)
|
||||
output_lines.append(f"> 🔊 {we.phonetic}")
|
||||
output_lines.append("")
|
||||
output_lines.append(md.heading("释义", 3))
|
||||
output_lines.append(md.bullet_list(we.definition))
|
||||
if we.example_sentence:
|
||||
output_lines.append("")
|
||||
output_lines.append(md.heading("例句", 3))
|
||||
output_lines.append(f"> {we.example_sentence}")
|
||||
|
||||
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.TRANSLATE and state.translated_text:
|
||||
output_lines.append(md.heading("🌐 翻译结果", 2))
|
||||
output_lines.append("")
|
||||
output_lines.append(md.heading("原文", 3))
|
||||
output_lines.append(f"> {state.source_text}")
|
||||
output_lines.append("")
|
||||
output_lines.append(md.heading("译文", 3))
|
||||
output_lines.append(f"> {state.translated_text}")
|
||||
|
||||
elif state.action == DictionaryAction.EXTRACT and state.extracted_terms:
|
||||
output_lines.append(md.heading("📝 提取的术语", 2))
|
||||
output_lines.append("")
|
||||
terms_data = [
|
||||
{"术语": t.term, "释义": t.definition, "分类": t.category}
|
||||
for t in state.extracted_terms
|
||||
]
|
||||
output_lines.append(md.table(terms_data))
|
||||
|
||||
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}")
|
||||
output_lines.append(md.heading("🌟 每日一词", 2))
|
||||
output_lines.append("")
|
||||
output_lines.append(md.heading(f"{dw.word}", 3))
|
||||
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
|
||||
output_lines.append(f"> 🔊 {dw.phonetic}")
|
||||
output_lines.append("")
|
||||
output_lines.append(md.bullet_list(dw.definition))
|
||||
|
||||
else:
|
||||
if not state.final_result:
|
||||
state.final_result = "词典操作完成"
|
||||
output_lines.append(md.heading("✨ 操作完成", 2))
|
||||
output_lines.append("您的请求已处理。")
|
||||
|
||||
# 页脚提示
|
||||
output_lines.append("")
|
||||
output_lines.append("---")
|
||||
output_lines.append("💡 提示：您可以继续查询其他单词、翻译文本，或者提取术语！")
|
||||
|
||||
state.final_result = "\n".join(output_lines)
|
||||
state.success = True
|
||||
state.current_phase = "completed"
|
||||
|
||||
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"
|
||||
|
||||
Reference in New Issue
Block a user