feat: 添加子图API端点和前端测试面板,包含确定取消继续交互
Some checks failed
构建并部署 AI Agent 服务 / deploy (push) Failing after 6m3s

This commit is contained in:
2026-04-25 19:38:22 +08:00
parent 96dc01f8c2
commit 3bc9b19bab
4 changed files with 524 additions and 0 deletions

View File

@@ -350,3 +350,235 @@ if __name__ == "__main__":
# 使用环境变量或默认端口 8079避免与 llama.cpp 的 8081 端口冲突)
port = int(BACKEND_PORT)
uvicorn.run(app, host="0.0.0.0", port=port)
# ==================== 子图专用 API 端点 ====================
# 简化版本,直接调用各个子图,无需完整 agent_service
# 注意:这些是独立测试用的简化端点,方便前端直接调用
@app.get("/subgraph/dictionary/{action}")
async def dictionary_subgraph_api(
action: str,
query: str = "",
user_id: str = "default"
):
"""词典子图简化 API"""
from backend.app.agent_subgraphs.dictionary import (
DictionaryState,
DictionaryAction,
parse_intent,
format_result
)
from backend.app.agent_subgraphs.dictionary.nodes import (
query_word, translate_text, extract_terms, get_daily_word
)
# 创建初始状态
state = DictionaryState(user_query=query, user_id=user_id)
# 处理 action
try:
if action == "query":
state.action = DictionaryAction.QUERY
state.action_params = {"word": query}
state = query_word(state)
elif action == "translate":
state.action = DictionaryAction.TRANSLATE
state.source_text = query
state = translate_text(state)
elif action == "daily":
state.action = DictionaryAction.DAILY_WORD
state = get_daily_word(state)
elif action == "extract":
state.action = DictionaryAction.EXTRACT
state.action_params = {"text": query}
state = extract_terms(state)
else:
# 自动解析意图
state = parse_intent(state)
# 根据解析后的 action 调用
if state.action == DictionaryAction.QUERY:
state = query_word(state)
elif state.action == DictionaryAction.TRANSLATE:
state = translate_text(state)
elif state.action == DictionaryAction.DAILY_WORD:
state = get_daily_word(state)
elif state.action == DictionaryAction.EXTRACT:
state = extract_terms(state)
# 格式化结果
state = format_result(state)
return {
"success": True,
"action": str(state.action),
"result": state.final_result,
"raw_data": {
"word_entry": vars(state.word_entry) if state.word_entry else None,
"translated_text": state.translated_text,
"extracted_terms": [vars(t) for t in state.extracted_terms],
"daily_word": vars(state.daily_word) if state.daily_word else None
}
}
except Exception as e:
return {"success": False, "error": str(e)}
@app.get("/subgraph/news/{action}")
async def news_subgraph_api(
action: str,
query: str = "",
user_id: str = "default"
):
"""资讯子图简化 API"""
from backend.app.agent_subgraphs.news_analysis import (
NewsAnalysisState,
NewsAction,
parse_intent,
format_result
)
from backend.app.agent_subgraphs.news_analysis.nodes import (
query_news, analyze_url, extract_keywords, generate_report
)
# 创建初始状态
state = NewsAnalysisState(user_query=query, user_id=user_id)
# 处理 action
try:
if action == "query":
state.action = NewsAction.QUERY_NEWS
state = query_news(state)
elif action == "analyze":
state.action = NewsAction.ANALYZE_URL
state.custom_urls = [query]
state = analyze_url(state)
elif action == "keywords":
state.action = NewsAction.EXTRACT_KEYWORDS
state = extract_keywords(state)
elif action == "report":
state.action = NewsAction.GENERATE_REPORT
state = generate_report(state)
else:
# 自动解析意图
state = parse_intent(state)
# 根据解析后的 action 调用
if state.action == NewsAction.QUERY_NEWS:
state = query_news(state)
elif state.action == NewsAction.ANALYZE_URL:
state.custom_urls = [query]
state = analyze_url(state)
elif state.action == NewsAction.EXTRACT_KEYWORDS:
state = extract_keywords(state)
elif state.action == NewsAction.GENERATE_REPORT:
state = generate_report(state)
# 格式化结果
state = format_result(state)
return {
"success": True,
"action": str(state.action),
"result": state.final_result,
"raw_data": {
"news_items": [vars(item) for item in state.news_items],
"extracted_keywords": state.extracted_keywords,
"report_content": state.report_content
}
}
except Exception as e:
return {"success": False, "error": str(e)}
@app.get("/subgraph/contact/{action}")
async def contact_subgraph_api(
action: str,
query: str = "",
user_id: str = "default"
):
"""通讯录子图简化 API"""
from backend.app.agent_subgraphs.contact import (
ContactState,
ContactAction,
parse_intent,
format_result
)
from backend.app.agent_subgraphs.contact.nodes import (
list_contacts, add_contact, list_emails, generate_email_draft, sniff_contacts
)
# 创建初始状态
state = ContactState(user_query=query, user_id=user_id)
# 处理 action
try:
if action == "list":
state.action = ContactAction.CONTACT_LIST
state = list_contacts(state)
elif action == "add":
state.action = ContactAction.CONTACT_ADD
state = add_contact(state)
elif action == "emails":
state.action = ContactAction.EMAIL_LIST
state = list_emails(state)
elif action == "draft":
state.action = ContactAction.EMAIL_SEND
state = generate_email_draft(state)
elif action == "sniff":
state.action = ContactAction.SNIFF_CONTACTS
state = sniff_contacts(state)
else:
# 自动解析意图
state = parse_intent(state)
# 根据解析后的 action 调用
if state.action == ContactAction.CONTACT_LIST:
state = list_contacts(state)
elif state.action == ContactAction.CONTACT_ADD:
state = add_contact(state)
elif state.action == ContactAction.EMAIL_LIST:
state = list_emails(state)
elif state.action == ContactAction.EMAIL_SEND:
state = generate_email_draft(state)
elif state.action == ContactAction.SNIFF_CONTACTS:
state = sniff_contacts(state)
# 格式化结果
state = format_result(state)
return {
"success": True,
"action": str(state.action),
"result": state.final_result,
"raw_data": {
"contacts": [vars(c) for c in state.contacts],
"emails": [vars(e) for e in state.emails],
"current_contact": vars(state.current_contact) if state.current_contact else None,
"draft": {
"subject": state.draft_subject,
"recipient": state.draft_recipient,
"body": state.draft_body
},
"sniffed": [vars(c) for c in state.sniffed_contacts]
}
}
except Exception as e:
return {"success": False, "error": str(e)}
@app.get("/subgraph/help")
async def subgraph_help_api():
"""子图 API 使用帮助"""
return {
"dictionary": {
"actions": ["query", "translate", "daily", "extract", "auto"],
"endpoint": "/subgraph/dictionary/{action}"
},
"news": {
"actions": ["query", "analyze", "keywords", "report", "auto"],
"endpoint": "/subgraph/news/{action}"
},
"contact": {
"actions": ["list", "add", "emails", "draft", "sniff", "auto"],
"endpoint": "/subgraph/contact/{action}"
}
}