root
e3adb45454
Some checks failed
构建并部署 AI Agent 服务 / deploy (push) Failing after 6m15s
- 更新 intent.py 为 React 模式推理器 - 新增 react_nodes.py: React 模式节点 - 新增 retry_utils.py: 超时和重试工具 - 更新 state.py: 支持循环步数和错误记录 - 重写 subgraph_builder.py: 完整 React 循环流程 - 结构化错误输出，符合 Agent 执行循环最佳实践 - 限制最大推理步数 ≤40，防止无限循环 - RAG 检索带重试和超时保护 - 子图错误可传递给主图处理
382 lines
14 KiB
Python
382 lines
14 KiB
Python
"""
|
||
意图理解与推理模块 (React模式)
|
||
Intent Understanding & Reasoning Module (React Pattern)
|
||
|
||
这个模块实现了 React (Reasoning + Acting) 模式的意图理解节点，用于：
|
||
1. 理解用户的查询意图
|
||
2. 判断是否需要调用 RAG 检索
|
||
3. 判断是否需要重新检索
|
||
4. 决定下一步的行动
|
||
5. 支持条件路由扩展
|
||
|
||
核心组件：
|
||
- ReasoningAction: 推理动作枚举
|
||
- ReasoningResult: 推理结果数据类
|
||
- ReactIntentReasoner: React 模式意图推理器
|
||
"""
|
||
|
||
import re
|
||
from typing import Dict, Any, Optional, List, Set, Tuple
|
||
from dataclasses import dataclass, field
|
||
from enum import Enum, auto
|
||
from abc import ABC, abstractmethod
|
||
|
||
|
||
class ReasoningAction(Enum):
|
||
"""推理动作枚举 - 决定下一步做什么"""
|
||
DIRECT_RESPONSE = auto() # 直接回答，不需要额外信息
|
||
RETRIEVE_RAG = auto() # 需要调用 RAG 检索
|
||
RERIEVE_RAG = auto() # 需要重新检索 (优化前版本，兼容保留)
|
||
RE_RETRIEVE_RAG = auto() # 需要重新检索 (修正拼写)
|
||
CALL_TOOL = auto() # 需要调用其他工具
|
||
CLARIFY = auto() # 需要澄清用户的问题
|
||
ROUTE_SUBGRAPH = auto() # 需要路由到子图
|
||
UNKNOWN = auto() # 未知动作
|
||
|
||
|
||
@dataclass
|
||
class RetrievalConfig:
|
||
"""检索配置"""
|
||
need_retrieval: bool = False # 是否需要检索
|
||
need_re_retrieval: bool = False # 是否需要重新检索
|
||
retrieval_query: Optional[str] = None # 优化后的检索查询
|
||
collection_name: Optional[str] = None # 检索的集合名称
|
||
k: int = 5 # 返回数量
|
||
score_threshold: float = 0.3 # 相似度阈值
|
||
metadata: Dict[str, Any] = field(default_factory=dict)
|
||
|
||
|
||
@dataclass
|
||
class ReasoningResult:
|
||
"""推理结果数据类"""
|
||
action: ReasoningAction = ReasoningAction.UNKNOWN # 决定的动作
|
||
confidence: float = 0.0 # 置信度
|
||
reasoning: str = "" # 推理过程说明
|
||
retrieval_config: RetrievalConfig = field(default_factory=RetrievalConfig)
|
||
extracted_entities: Dict[str, Any] = field(default_factory=dict) # 提取的实体
|
||
next_hints: List[str] = field(default_factory=list) # 下一步提示
|
||
original_query: str = "" # 原始查询
|
||
metadata: Dict[str, Any] = field(default_factory=dict)
|
||
|
||
|
||
class BaseIntentReasoner(ABC):
|
||
"""意图推理器基类"""
|
||
|
||
@abstractmethod
|
||
def reason(
|
||
self,
|
||
query: str,
|
||
context: Optional[Dict[str, Any]] = None
|
||
) -> ReasoningResult:
|
||
"""
|
||
推理意图，决定下一步动作
|
||
|
||
Args:
|
||
query: 用户查询
|
||
context: 上下文信息，可能包括：
|
||
- messages: 对话历史
|
||
- retrieved_docs: 已检索的文档
|
||
- previous_actions: 之前的动作
|
||
- user_id: 用户ID
|
||
- etc.
|
||
|
||
Returns:
|
||
ReasoningResult: 推理结果
|
||
"""
|
||
pass
|
||
|
||
|
||
class RuleBasedReactReasoner(BaseIntentReasoner):
|
||
"""基于规则的 React 推理器"""
|
||
|
||
def __init__(self):
|
||
# 检索触发关键词
|
||
self._retrieval_keywords = {
|
||
"什么", "怎么", "如何", "为什么", "哪", "谁", "多少",
|
||
"介绍", "解释", "说明", "资料", "文档", "查询", "搜索",
|
||
"find", "search", "what", "how", "why", "where", "who",
|
||
"tell me", "explain", "about", "information"
|
||
}
|
||
|
||
# 重新检索触发关键词
|
||
self._re_retrieval_keywords = {
|
||
"再", "重新", "更多", "不够", "不足", "其他", "另外",
|
||
"没找到", "找不到", "没有", "不对", "不是",
|
||
"again", "more", "another", "other", "didn't find", "not enough"
|
||
}
|
||
|
||
# 澄清触发关键词
|
||
self._clarify_keywords = {
|
||
"?", "？", "哪个", "哪些", "哪位", "什么意思",
|
||
"请问", "能详细", "具体点", "举个例子"
|
||
}
|
||
|
||
# 工具调用关键词
|
||
self._tool_keywords = {
|
||
"天气", "weather", "邮件", "email", "联系人", "contact",
|
||
"翻译", "translate", "词典", "dictionary"
|
||
}
|
||
|
||
# 子图路由关键词映射
|
||
self._subgraph_keywords = {
|
||
"contact": {"通讯录", "联系人", "contact", "email", "邮件"},
|
||
"dictionary": {"词典", "单词", "翻译", "dictionary", "translate"},
|
||
"news_analysis": {"资讯", "新闻", "分析", "news", "report"},
|
||
}
|
||
|
||
# 直接回答模式（问候、感谢等）
|
||
self._direct_response_patterns = [
|
||
(r'^(你好|您好|hi|hello|hey|早上好|下午好|晚上好|哈喽)', ReasoningAction.DIRECT_RESPONSE),
|
||
(r'^(谢谢|感谢|多谢|thanks|thank you)', ReasoningAction.DIRECT_RESPONSE),
|
||
(r'^(再见|拜拜|bye|goodbye|回见)', ReasoningAction.DIRECT_RESPONSE),
|
||
]
|
||
|
||
def reason(
|
||
self,
|
||
query: str,
|
||
context: Optional[Dict[str, Any]] = None
|
||
) -> ReasoningResult:
|
||
"""
|
||
基于规则的推理
|
||
"""
|
||
context = context or {}
|
||
query_lower = query.lower()
|
||
result = ReasoningResult(original_query=query)
|
||
|
||
# 1. 先检查是否是直接回答模式
|
||
for pattern, action in self._direct_response_patterns:
|
||
if re.match(pattern, query, re.IGNORECASE):
|
||
result.action = action
|
||
result.confidence = 0.95
|
||
result.reasoning = "检测到问候、感谢或告别语，直接回答"
|
||
return result
|
||
|
||
# 2. 检查是否需要路由到子图（优先级高于重新检索，避免"有没有"误触发）
|
||
for subgraph, keywords in self._subgraph_keywords.items():
|
||
if any(kw in query_lower for kw in keywords):
|
||
result.action = ReasoningAction.ROUTE_SUBGRAPH
|
||
result.confidence = 0.9
|
||
result.reasoning = f"检测到 {subgraph} 子图意图"
|
||
result.metadata["target_subgraph"] = subgraph
|
||
return result
|
||
|
||
# 3. 检查是否需要重新检索
|
||
has_re_retrieval = any(kw in query_lower for kw in self._re_retrieval_keywords)
|
||
# 同时检查上下文中是否有之前的检索结果但不够好
|
||
previous_retrieval = context.get("retrieved_docs")
|
||
if has_re_retrieval or (previous_retrieval and len(previous_retrieval) < 2):
|
||
result.action = ReasoningAction.RE_RETRIEVE_RAG
|
||
result.confidence = 0.85 if has_re_retrieval else 0.7
|
||
result.reasoning = "检测到需要重新检索的意图"
|
||
result.retrieval_config = RetrievalConfig(
|
||
need_retrieval=True,
|
||
need_re_retrieval=True,
|
||
retrieval_query=self._optimize_retrieval_query(query),
|
||
k=10 # 重新检索时返回更多结果
|
||
)
|
||
return result
|
||
|
||
# 4. 检查是否需要调用工具
|
||
has_tool = any(kw in query_lower for kw in self._tool_keywords)
|
||
if has_tool:
|
||
result.action = ReasoningAction.CALL_TOOL
|
||
result.confidence = 0.8
|
||
result.reasoning = "检测到工具调用意图"
|
||
return result
|
||
|
||
# 5. 检查是否需要澄清
|
||
has_clarify = any(kw in query_lower for kw in self._clarify_keywords)
|
||
# 或者查询太短、太模糊
|
||
if has_clarify or len(query.strip()) < 3:
|
||
result.action = ReasoningAction.CLARIFY
|
||
result.confidence = 0.75
|
||
result.reasoning = "检测到需要澄清的意图"
|
||
result.next_hints = [
|
||
"请提供更多细节",
|
||
"您想了解什么方面的内容？",
|
||
"能否具体说明一下？"
|
||
]
|
||
return result
|
||
|
||
# 6. 检查是否需要 RAG 检索
|
||
has_retrieval = any(kw in query_lower for kw in self._retrieval_keywords)
|
||
if has_retrieval or len(query.strip()) > 5:
|
||
result.action = ReasoningAction.RETRIEVE_RAG
|
||
result.confidence = 0.85 if has_retrieval else 0.6
|
||
result.reasoning = "检测到需要检索知识库的意图"
|
||
result.retrieval_config = RetrievalConfig(
|
||
need_retrieval=True,
|
||
retrieval_query=self._optimize_retrieval_query(query),
|
||
k=5
|
||
)
|
||
return result
|
||
|
||
# 7. 默认直接回答
|
||
result.action = ReasoningAction.DIRECT_RESPONSE
|
||
result.confidence = 0.6
|
||
result.reasoning = "默认直接回答模式"
|
||
return result
|
||
|
||
def _optimize_retrieval_query(self, query: str) -> str:
|
||
"""优化检索查询，去掉不必要的语气词"""
|
||
# 去掉常见的前缀
|
||
prefixes_to_remove = [
|
||
"请告诉我", "帮我查一下", "我想知道", "能不能告诉我",
|
||
"请问", "你知道", "帮我找", "搜索一下", "查询一下"
|
||
]
|
||
optimized = query
|
||
for prefix in prefixes_to_remove:
|
||
if optimized.startswith(prefix):
|
||
optimized = optimized[len(prefix):]
|
||
|
||
# 去掉常见的后缀
|
||
suffixes_to_remove = ["吗？", "呢？", "吧？", "吗", "呢", "吧", "？", "?"]
|
||
for suffix in suffixes_to_remove:
|
||
if optimized.endswith(suffix):
|
||
optimized = optimized[:-len(suffix)]
|
||
|
||
return optimized.strip()
|
||
|
||
|
||
class LLMReactReasoner(BaseIntentReasoner):
|
||
"""
|
||
基于 LLM 的 React 推理器
|
||
使用大语言模型进行更智能的推理判断
|
||
"""
|
||
|
||
def __init__(self, llm_client=None):
|
||
"""
|
||
初始化 LLM 推理器
|
||
|
||
Args:
|
||
llm_client: LLM 客户端，需要支持调用方法
|
||
"""
|
||
self.llm_client = llm_client
|
||
self.rule_based = RuleBasedReactReasoner()
|
||
|
||
def reason(
|
||
self,
|
||
query: str,
|
||
context: Optional[Dict[str, Any]] = None
|
||
) -> ReasoningResult:
|
||
"""
|
||
使用 LLM 进行推理，失败时回退到规则推理
|
||
"""
|
||
try:
|
||
if self.llm_client:
|
||
return self._reason_with_llm(query, context)
|
||
except Exception:
|
||
pass
|
||
|
||
# LLM 不可用或失败，回退到规则推理
|
||
return self.rule_based.reason(query, context)
|
||
|
||
def _reason_with_llm(
|
||
self,
|
||
query: str,
|
||
context: Optional[Dict[str, Any]] = None
|
||
) -> ReasoningResult:
|
||
"""
|
||
使用 LLM 进行推理（需要实现具体的 LLM 调用逻辑）
|
||
"""
|
||
# 这里是一个示例实现，实际项目需要连接真实的 LLM
|
||
prompt = self._build_reasoning_prompt(query, context)
|
||
|
||
# 模拟 LLM 返回（实际项目中替换为真实调用）
|
||
# 这里我们还是先调用规则推理作为示例
|
||
return self.rule_based.reason(query, context)
|
||
|
||
def _build_reasoning_prompt(self, query: str, context: Optional[Dict[str, Any]]) -> str:
|
||
"""构建推理提示词"""
|
||
context_str = ""
|
||
if context:
|
||
context_lines = []
|
||
if "messages" in context:
|
||
context_lines.append(f"对话历史: {len(context['messages'])} 条")
|
||
if "retrieved_docs" in context:
|
||
context_lines.append(f"已检索文档: {len(context['retrieved_docs'])} 条")
|
||
context_str = "\n".join(context_lines)
|
||
|
||
return f"""你是一个意图推理助手，需要判断用户的查询应该如何处理。
|
||
|
||
用户查询: {query}
|
||
|
||
上下文信息:
|
||
{context_str or '无额外上下文'}
|
||
|
||
请判断下一步应该做什么，可选动作:
|
||
1. DIRECT_RESPONSE - 直接回答，不需要额外信息
|
||
2. RETRIEVE_RAG - 需要调用知识库检索
|
||
3. RE_RETRIEVE_RAG - 需要重新检索更多/更好的结果
|
||
4. CALL_TOOL - 需要调用其他工具
|
||
5. CLARIFY - 需要澄清用户的问题
|
||
6. ROUTE_SUBGRAPH - 需要路由到子图
|
||
|
||
请以 JSON 格式输出你的判断。
|
||
"""
|
||
|
||
|
||
def create_react_reasoner(
|
||
use_llm: bool = False,
|
||
llm_client=None
|
||
) -> BaseIntentReasoner:
|
||
"""
|
||
创建 React 模式意图推理器工厂函数
|
||
|
||
Args:
|
||
use_llm: 是否使用 LLM 推理
|
||
llm_client: LLM 客户端实例
|
||
|
||
Returns:
|
||
BaseIntentReasoner: 推理器实例
|
||
"""
|
||
if use_llm:
|
||
return LLMReactReasoner(llm_client)
|
||
return RuleBasedReactReasoner()
|
||
|
||
|
||
# 便捷函数 - 直接推理
|
||
def react_reason(
|
||
query: str,
|
||
context: Optional[Dict[str, Any]] = None,
|
||
reasoner: Optional[BaseIntentReasoner] = None
|
||
) -> ReasoningResult:
|
||
"""
|
||
便捷函数：直接进行 React 推理
|
||
|
||
Args:
|
||
query: 用户查询
|
||
context: 上下文信息
|
||
reasoner: 可选的推理器实例
|
||
|
||
Returns:
|
||
ReasoningResult: 推理结果
|
||
"""
|
||
if reasoner is None:
|
||
reasoner = create_react_reasoner()
|
||
return reasoner.reason(query, context)
|
||
|
||
|
||
# 条件路由辅助函数
|
||
def get_route_by_reasoning(result: ReasoningResult) -> str:
|
||
"""
|
||
根据推理结果获取路由字符串
|
||
|
||
Args:
|
||
result: 推理结果
|
||
|
||
Returns:
|
||
str: 路由标识
|
||
"""
|
||
action_to_route = {
|
||
ReasoningAction.DIRECT_RESPONSE: "direct_response",
|
||
ReasoningAction.RETRIEVE_RAG: "retrieve_rag",
|
||
ReasoningAction.RE_RETRIEVE_RAG: "re_retrieve_rag",
|
||
ReasoningAction.RERIEVE_RAG: "re_retrieve_rag", # 兼容旧拼写
|
||
ReasoningAction.CALL_TOOL: "call_tool",
|
||
ReasoningAction.CLARIFY: "clarify",
|
||
ReasoningAction.ROUTE_SUBGRAPH: result.metadata.get("target_subgraph", "unknown_subgraph"),
|
||
ReasoningAction.UNKNOWN: "unknown",
|
||
}
|
||
return action_to_route.get(result.action, "unknown")
|