root/ailine
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

给关键节点添加思考过程输出
All checks were successful
构建并部署 AI Agent 服务 / deploy (push) Successful in 6m41s
Details

Browse Source 
- react_reason_node: 直接发送自定义推理事件
- web_search_node: 添加开始/完成/错误事件
- rag_retrieve_node: 添加开始/完成/重试/错误事件
- 子图包装器: 添加子图开始/完成/错误事件
This commit is contained in:
root
2026-05-02 09:23:07 +08:00
parent afddea61f8
commit 3f6bbdec92
3 changed files with 198 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

81
backend/app/main_graph/nodes/rag_nodes.py
View File

@@ -159,18 +159,38 @@ def _rag_retrieve_core(state: MainGraphState) -> MainGraphState:
# ========== RAG 检索节点（带超时和重试）==========
def rag_retrieve_node(state: MainGraphState) -> MainGraphState:
async def rag_retrieve_node(state: MainGraphState, config: Optional[Dict[str, Any]] = None) -> MainGraphState:
"""
RAG 检索节点：带超时和重试，真正利用已有 RAG 代码
Args:
state: 主图状态
config: LangChain 配置
Returns:
更新后的状态
"""
state.current_phase = "rag_retrieving"
# 发送开始事件
if config:
try:
from langchain_core.callbacks.manager import adispatch_custom_event
callbacks = config.get("callbacks")
if callbacks:
await adispatch_custom_event(
"react_reasoning",
{
"step": state.reasoning_step,
"action": "rag_retrieve_start",
"confidence": 1.0,
"reasoning": "开始执行 RAG 检索..."
},
callbacks=callbacks
)
except Exception as e:
info(f"[rag_retrieve_node] 无法发送开始事件: {e}")
start_time = time.time()
last_error = None
@@ -185,6 +205,27 @@ def rag_retrieve_node(state: MainGraphState) -> MainGraphState:
"success": True,
"time": time.time() - start_time
}
# 发送完成事件
if config:
try:
from langchain_core.callbacks.manager import adispatch_custom_event
callbacks = config.get("callbacks")
if callbacks:
doc_count = len(result.rag_docs) if result.rag_docs else 0
await adispatch_custom_event(
"react_reasoning",
{
"step": state.reasoning_step,
"action": "rag_retrieve_complete",
"confidence": 1.0,
"reasoning": f"RAG 检索完成，找到 {doc_count} 条相关文档"
},
callbacks=callbacks
)
except Exception as e:
info(f"[rag_retrieve_node] 无法发送完成事件: {e}")
return result
except Exception as e:
@@ -193,6 +234,25 @@ def rag_retrieve_node(state: MainGraphState) -> MainGraphState:
if attempt >= RAG_RETRY_CONFIG.max_retries:
break
# 发送重试事件
if config:
try:
from langchain_core.callbacks.manager import adispatch_custom_event
callbacks = config.get("callbacks")
if callbacks:
await adispatch_custom_event(
"react_reasoning",
{
"step": state.reasoning_step,
"action": "rag_retrieve_retry",
"confidence": 1.0,
"reasoning": f"RAG 检索失败，第 {attempt + 1} 次重试..."
},
callbacks=callbacks
)
except Exception as e:
info(f"[rag_retrieve_node] 无法发送重试事件: {e}")
# 指数退避等待
delay = RAG_RETRY_CONFIG.base_delay * (2 ** attempt)
time.sleep(min(delay, RAG_RETRY_CONFIG.max_delay))
@@ -219,6 +279,25 @@ def rag_retrieve_node(state: MainGraphState) -> MainGraphState:
state.current_error = error_record
state.current_phase = "error_handling"
# 发送错误事件
if config:
try:
from langchain_core.callbacks.manager import adispatch_custom_event
callbacks = config.get("callbacks")
if callbacks:
await adispatch_custom_event(
"react_reasoning",
{
"step": state.reasoning_step,
"action": "rag_retrieve_error",
"confidence": 1.0,
"reasoning": f"RAG 检索失败: {str(last_error)}"
},
callbacks=callbacks
)
except Exception as e:
info(f"[rag_retrieve_node] 无法发送错误事件: {e}")
return state

59
backend/app/main_graph/nodes/react_nodes.py
View File

@@ -121,12 +121,31 @@ async def react_reason_node(state: MainGraphState, config: Optional[Dict[str, An
# ========== 2. 联网搜索节点 ==========
def web_search_node(state: MainGraphState) -> MainGraphState:
async def web_search_node(state: MainGraphState, config: Optional[Dict[str, Any]] = None) -> MainGraphState:
"""
联网搜索节点：执行搜索并将结果保存到状态
"""
state.current_phase = "web_searching"
# 发送开始事件
if config:
try:
from langchain_core.callbacks.manager import adispatch_custom_event
callbacks = config.get("callbacks")
if callbacks:
await adispatch_custom_event(
"react_reasoning",
{
"step": state.reasoning_step,
"action": "web_search_start",
"confidence": 1.0,
"reasoning": "开始执行联网搜索..."
},
callbacks=callbacks
)
except Exception as e:
info(f"[web_search_node] 无法发送开始事件: {e}")
# 获取搜索查询
reasoning_result = state.debug_info.get("reasoning_result")
search_query = reasoning_result.metadata.get("search_query", state.user_query) if reasoning_result else state.user_query
@@ -151,6 +170,25 @@ def web_search_node(state: MainGraphState) -> MainGraphState:
state.success = True
print(f"[WebSearch] 搜索完成")
# 发送完成事件
if config:
try:
from langchain_core.callbacks.manager import adispatch_custom_event
callbacks = config.get("callbacks")
if callbacks:
await adispatch_custom_event(
"react_reasoning",
{
"step": state.reasoning_step,
"action": "web_search_complete",
"confidence": 1.0,
"reasoning": f"联网搜索完成，找到 {len(search_result) if isinstance(search_result, list) else 1} 条结果"
},
callbacks=callbacks
)
except Exception as e:
info(f"[web_search_node] 无法发送完成事件: {e}")
except Exception as e:
from app.main_graph.state import ErrorRecord, ErrorSeverity
from datetime import datetime
@@ -169,6 +207,25 @@ def web_search_node(state: MainGraphState) -> MainGraphState:
state.current_error = error_record
state.current_phase = "error_handling"
state.success = False
# 发送错误事件
if config:
try:
from langchain_core.callbacks.manager import adispatch_custom_event
callbacks = config.get("callbacks")
if callbacks:
await adispatch_custom_event(
"react_reasoning",
{
"step": state.reasoning_step,
"action": "web_search_error",
"confidence": 1.0,
"reasoning": f"联网搜索失败: {str(e)}"
},
callbacks=callbacks
)
except Exception as e:
info(f"[web_search_node] 无法发送错误事件: {e}")
return state