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

修复三个问题：1. 子图执行后的无限循环 2. llm_call没有输出 3. 思考打印两次

Browse Source 
- 子图执行后直接进入finalize，避免回到react_reason循环
- llm_call节点检查是否已有final_result，避免重复调用LLM
- 直接在react_reason_node中通过adispatch_custom_event发送推理事件，避免通过state传递导致重复
This commit is contained in:
root
2026-05-02 09:00:34 +08:00
parent bd2c20d927
commit 2893accbc4
5 changed files with 139 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files

65
backend/app/agent/agent_service.py
View File

@@ -146,7 +146,7 @@ class AIAgentService:
"current_action": CurrentAction.NONE
}
# ========== 意图识别（保留用于日志） ==========
# ========== 意图识别（保留用于日志）==========
intent_result = await self.intent_classifier.classify(message)
info(f"🧠 意图识别: {intent_result.intent_type} (置信度: {intent_result.confidence:.2f})")
info(f"📝 推理: {intent_result.reasoning}")
@@ -265,34 +265,9 @@ class AIAgentService:
updates_data = chunk["data"]
serialized_data = self._serialize_value(updates_data)
# 检查是否有最新的推理内容（来自 react_reason_node）
if isinstance(serialized_data, dict):
# 遍历所有节点的数据
for node_name, node_data in serialized_data.items():
if isinstance(node_data, dict) and "debug_info" in node_data:
debug_info = node_data["debug_info"]
latest_reasoning = debug_info.get("latest_reasoning")
if latest_reasoning and not latest_reasoning.get("sent"):
# 发送推理过程到前端
step = latest_reasoning.get("step", 1)
action = latest_reasoning.get("action", "unknown")
confidence = latest_reasoning.get("confidence", 0)
reasoning = latest_reasoning.get("reasoning", "")
info(f"[Agent Service] 发送推理过程 #{step}: {action}")
# 发送推理事件
yield {
"type": "react_reasoning",
"step": step,
"action": action,
"confidence": confidence,
"reasoning": reasoning
}
# 标记为已发送（避免重复发送）
latest_reasoning["sent"] = True
# 关键修复：不再从 updates 中读取 latest_reasoning，避免重复
# 因为我们现在直接通过 custom 事件发送推理结果了
# 检查是否有人工审核请求
if "review_pending" in serialized_data and serialized_data["review_pending"]:
review_id = serialized_data.get("review_id", "")
@@ -302,7 +277,7 @@ class AIAgentService:
"review_id": review_id,
"content": content_to_review
}
# 检查是否有工具结果
if "messages" in serialized_data:
for msg in serialized_data["messages"]:
@@ -311,7 +286,7 @@ class AIAgentService:
tool_call_id = msg.get("tool_call_id", "")
tool_name = msg.get("name", "")
tool_output = msg.get("content", "")
if tool_call_id in tool_calls_in_progress:
yield {
"type": "tool_call_end",
@@ -320,19 +295,33 @@ class AIAgentService:
"result": tool_output
}
del tool_calls_in_progress[tool_call_id]
processed_event = {
"type": "state_update",
"data": serialized_data
}
elif chunk_type == "custom":
info(f"🎯 处理custom chunk")
serialized_data = self._serialize_value(chunk["data"])
processed_event = {
"type": "custom",
"data": serialized_data
}
info(f"🎯 处理custom chunk: {chunk['data']}")
custom_data = chunk["data"]
# 关键修复：处理我们从 react_reason_node 发送的自定义推理事件
if isinstance(custom_data, dict) and custom_data.get("type") == "react_reasoning":
info(f"[Agent Service] 收到自定义推理事件")
yield {
"type": "react_reasoning",
"step": custom_data.get("step", 1),
"action": custom_data.get("action", "unknown"),
"confidence": custom_data.get("confidence", 0),
"reasoning": custom_data.get("reasoning", "")
}
else:
# 处理其他自定义事件
serialized_data = self._serialize_value(custom_data)
processed_event = {
"type": "custom",
"data": serialized_data
}
if processed_event:
yield processed_event

23
backend/app/core/intent.py
View File

@@ -107,8 +107,27 @@ class ReactIntentReasoner:
"""
context = context or {}
result = ReasoningResult(original_query=query)
# 关键修复 1：检查是否已经有检索结果或子图结果，如果是，直接回答
previous_actions = context.get("previous_actions", [])
if "subgraph_completed" in previous_actions:
result.action = ReasoningAction.DIRECT_RESPONSE
result.confidence = 1.0
result.reasoning = "子图已执行完成，直接回答"
return result
retrieved_docs = context.get("retrieved_docs", [])
messages = context.get("messages", [])
# 关键修复 2：如果已经有 rag_context 或 web_search_results（通过 messages 推断），直接回答
# 检查是否已经执行过 rag_retrieve 或 web_search
if "rag_retrieve" in previous_actions or "web_search" in previous_actions:
result.action = ReasoningAction.DIRECT_RESPONSE
result.confidence = 0.95
result.reasoning = "已获取信息，直接回答"
return result
# 策略1: 尝试使用 LLM 推理
# 策略1：尝试使用 LLM 推理
try:
llm_result = await self._reason_with_llm(query, context)
if llm_result.confidence >= 0.6: # 置信度足够高，直接返回
@@ -116,7 +135,7 @@ class ReactIntentReasoner:
except Exception as e:
print(f"[ReactReasoner] LLM 推理失败: {e}, 回退到规则")
# 策略2: LLM 失败或置信度低，使用规则匹配
# 策略2：LLM 失败或置信度低，使用规则匹配
return self._reason_with_rules(query, context)
async def _reason_with_llm(

17
backend/app/main_graph/nodes/llm_call.py
View File

@@ -37,11 +37,11 @@ def create_llm_call_node(llm, tools: list):
async def call_llm(state: MainGraphState, config: RunnableConfig) -> Dict[str, Any]:
"""
LLM 调用节点（异步方法）
Args:
state: 当前对话状态
config: LangChain/LangGraph 自动注入的配置，包含 callbacks 等信息
Returns:
更新后的状态字典
"""
@@ -49,6 +49,19 @@ def create_llm_call_node(llm, tools: list):
memory_context = getattr(state, "memory_context", "暂无用户信息")
start_time = time.time()
# 关键修复：如果 state.final_result 已经存在（比如子图执行完），直接返回
if state.final_result:
info(f"[llm_call] 检测到已有最终结果，直接返回: {state.final_result[:100]}...")
elapsed_time = time.time() - start_time
return {
"final_result": state.final_result,
"success": True,
"current_phase": "done",
"llm_calls": getattr(state, 'llm_calls', 0) + 1,
"last_elapsed_time": elapsed_time,
"turns_since_last_summary": getattr(state, 'turns_since_last_summary', 0) + 1,
}
try:
# 添加 RAG 上下文到消息

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

@@ -16,6 +16,7 @@ from datetime import datetime
# 导入我们的 intent.py
from app.core.intent import (
react_reason,
react_reason_async,
get_route_by_reasoning,
ReasoningAction,
ReasoningResult
@@ -31,31 +32,17 @@ from app.logger import info
# ========== 1. React 推理节点 ==========
def react_reason_node(state: MainGraphState, config: Optional[Dict[str, Any]] = None) -> MainGraphState:
async def react_reason_node(state: MainGraphState, config: Optional[Dict[str, Any]] = None) -> MainGraphState:
"""
React 模式推理节点：判断下一步做什么
React 模式推理节点：判断下一步做什么（异步版本）
Returns: 更新后的状态
"""
state.current_phase = "react_reasoning"
state.reasoning_step += 1
# 发送推理开始事件
if config:
try:
from langchain_core.runnables.config import RunnableConfig
# 尝试获取回调管理器并发送自定义事件
callbacks = config.get("callbacks")
if callbacks:
from langchain_core.callbacks.manager import adispatch_custom_event
# 注意：这是异步操作，我们需要特殊处理
# 这里我们使用自定义流式写入器（如果存在）
pass
except Exception as e:
info(f"[react_reason] 无法发送回调事件: {e}")
info(f"[react_reason] 第 {state.reasoning_step} 次推理开始")
# 检查是否超过最大步数
if state.reasoning_step > state.max_steps:
state.current_phase = "max_steps_exceeded"
@@ -66,7 +53,7 @@ def react_reason_node(state: MainGraphState, config: Optional[Dict[str, Any]] =
)
state.success = False
return state
# 准备上下文
context = {
"retrieved_docs": state.rag_docs,
@@ -74,15 +61,35 @@ def react_reason_node(state: MainGraphState, config: Optional[Dict[str, Any]] =
"messages": state.messages,
"errors": state.errors
}
# 使用 intent.py 进行推理
# 注意：这里使用同步版本，内部会根据情况处理
result: ReasoningResult = react_reason(state.user_query, context)
# 使用 intent.py 进行推理（现在直接用异步版本）
result: ReasoningResult = await react_reason_async(state.user_query, context)
info(f"[react_reason] 推理结果: action={result.action.name}, confidence={result.confidence}")
if result.reasoning:
info(f"[react_reason] 推理过程: {result.reasoning}")
# 关键修复：直接发送自定义事件给 agent_service，而不是通过 state
if config:
try:
from langchain_core.callbacks.manager import adispatch_custom_event
callbacks = config.get("callbacks")
if callbacks:
info(f"[react_reason] 直接发送推理事件 #{state.reasoning_step}")
await adispatch_custom_event(
"react_reasoning",
{
"step": state.reasoning_step,
"action": result.action.name,
"confidence": result.confidence,
"reasoning": result.reasoning
},
callbacks=callbacks
)
except Exception as e:
info(f"[react_reason] 无法发送自定义事件: {e}")
# 记录推理历史
state.reasoning_history.append({
"step": state.reasoning_step,
@@ -91,30 +98,24 @@ def react_reason_node(state: MainGraphState, config: Optional[Dict[str, Any]] =
"reasoning": result.reasoning,
"timestamp": datetime.now().isoformat()
})
# 更新状态
state.debug_info["last_reasoning"] = {
"action": result.action.name,
"confidence": result.confidence,
"reasoning": result.reasoning
}
# 保存推理结果到状态
state.debug_info["reasoning_result"] = result
# 确定下一步动作
state.last_action = result.action.name
# 发送推理完成事件（通过状态更新，agent_service 会处理）
# 我们在状态中保存推理内容，以便 agent_service 可以通过 state_update 事件发送
state.debug_info["latest_reasoning"] = {
"step": state.reasoning_step,
"action": result.action.name,
"confidence": result.confidence,
"reasoning": result.reasoning,
"sent": False # 标记是否已发送到前端
}
# 关键修复：不再设置 latest_reasoning，避免 agent_service 重复读取
if "latest_reasoning" in state.debug_info:
del state.debug_info["latest_reasoning"]
return state
@@ -300,6 +301,16 @@ def route_by_reasoning(state: MainGraphState) -> str:
if state.retry_action and "rag" in state.retry_action.lower():
return "rag_retrieve"
return "react_reason"
# 关键修复：检查是否已经执行过子图，如果是，直接去 llm_call
previous_actions = [h.get("action") for h in state.reasoning_history]
if "subgraph_completed" in previous_actions or state.final_result:
return "llm_call"
# 检查是否刚刚执行完 rag 或 web search，应该继续推理一次然后去 llm_call
# 但为了避免死循环，我们设置一个简单的规则
if len(previous_actions) > 3:
return "llm_call"
# 获取推理结果
reasoning_result: Optional[ReasoningResult] = state.debug_info.get("reasoning_result")

37
backend/app/main_graph/utils/main_graph_builder.py
View File

@@ -61,15 +61,34 @@ def wrap_subgraph_for_error_handling(subgraph, name: str):
result = subgraph.invoke(state)
# 更新主图状态
subgraph_result = None
if name == "contact":
state.contact_result = result
subgraph_result = result.get("final_result", "")
elif name == "dictionary":
state.dictionary_result = result
subgraph_result = result.get("final_result", "")
elif name == "news_analysis":
state.news_result = result
subgraph_result = result.get("final_result", "")
# 关键：设置最终结果，这样就不需要再回到 react_reason 了
if subgraph_result:
state.final_result = subgraph_result
else:
state.final_result = "子图执行完成"
# 标记成功
state.success = True
state.current_phase = "done"
# 标记不再需要推理，避免循环
state.reasoning_history.append({
"step": state.reasoning_step,
"action": "subgraph_completed",
"confidence": 1.0,
"reasoning": f"{name}子图执行完成",
"timestamp": datetime.now().isoformat()
})
return state
except Exception as e:
@@ -189,18 +208,18 @@ def build_react_main_graph(llm=None, tools=None, mem0_client=None) -> StateGraph
graph.add_node("finalize", finalize_node)
# ========== 添加边 ==========
# 第一阶段：记忆检索
if retrieve_memory_node:
graph.add_edge(START, "retrieve_memory")
graph.add_edge("retrieve_memory", "memory_trigger")
else:
graph.add_edge(START, "memory_trigger")
# 进入第二阶段
graph.add_edge("memory_trigger", "init_state")
graph.add_edge("init_state", "react_reason")
# 第二阶段：React 循环推理
graph.add_conditional_edges(
"react_reason",
@@ -215,15 +234,17 @@ def build_react_main_graph(llm=None, tools=None, mem0_client=None) -> StateGraph
"llm_call": "llm_call"
}
)
# 循环边
# 循环边（rag、web_search、error 回到 reason）
graph.add_edge("rag_retrieve", "react_reason")
graph.add_edge("web_search", "react_reason")
graph.add_edge("contact_subgraph", "react_reason")
graph.add_edge("dictionary_subgraph", "react_reason")
graph.add_edge("news_analysis_subgraph", "react_reason")
graph.add_edge("handle_error", "react_reason")
# 关键修改：子图执行完后直接去 finalize（避免循环）
graph.add_edge("contact_subgraph", "finalize")
graph.add_edge("dictionary_subgraph", "finalize")
graph.add_edge("news_analysis_subgraph", "finalize")
# 第三阶段：llm_call 后进入完成处理
if llm_node is not None:
if summarize_node: