DESKTOP-72TV0V4\caoxiaozhu
0e0f988264
- 新增 intent_router.py 意图路由模块 - 优化 context.py 上下文管理 - 增强 loop.py Agent 运行循环 - 更新 memory.py 记忆模块 - 修复 builtin.py 工具函数 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
279 lines
8.8 KiB
Python
279 lines
8.8 KiB
Python
"""Intent recognition system for routing user requests."""
|
|
|
|
import json
|
|
import logging
|
|
from enum import Enum
|
|
from typing import Any
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class IntentType(Enum):
|
|
"""Types of user intents."""
|
|
SIMPLE = "simple" # Simple Q&A, no tools needed
|
|
TOOL = "tool" # Needs tools (search, code, files, etc.)
|
|
SKILL = "skill" # Needs specific domain skill
|
|
TEAM = "team" # Needs multi-agent collaboration
|
|
UNKNOWN = "unknown" # Cannot determine
|
|
|
|
|
|
# Intent recognition prompt template
|
|
INTENT_PROMPT = """Analyze the user's message and classify their intent.
|
|
|
|
Intent Types:
|
|
- simple: General knowledge questions, greetings, casual conversation, simple Q&A
|
|
Examples: "你好", "介绍一下武汉", "什么是AI", "今天天气怎么样"
|
|
- tool: Requires external tools - web search, code execution, file operations, calculations
|
|
Examples: "搜索最新的AI新闻", "帮我运行这段代码", "读取文件内容", "计算这个表达式"
|
|
- skill: Requires specific domain skill (coding, design, analysis, etc.)
|
|
Examples: "用Python写一个排序算法", "分析这段代码的性能", "创建一个网页"
|
|
- team: Requires multiple agents working together
|
|
Examples: "让设计agent和开发agent一起完成这个任务", "创建一个团队来完成这个项目"
|
|
|
|
Guidelines:
|
|
- For greetings and simple questions, prefer "simple"
|
|
- Only use "tool" when user explicitly asks for search, execution, or file operations
|
|
- "introduce Wuhan" in Chinese is general knowledge - prefer "simple" unless user specifically asks for latest/current information
|
|
- If ambiguous, prefer "simple" to avoid unnecessary tool calls
|
|
|
|
User message: {message}
|
|
|
|
Respond with only the intent type (simple/tool/skill/team), no explanation:"""
|
|
|
|
|
|
class IntentRecognizer:
|
|
"""Recognizes user intent to route requests appropriately."""
|
|
|
|
def __init__(self, llm_provider=None):
|
|
"""Initialize intent recognizer.
|
|
|
|
Args:
|
|
llm_provider: LLM provider for intent recognition
|
|
"""
|
|
self._llm_provider = llm_provider
|
|
self._cache = {} # Simple cache for recent intents
|
|
|
|
def recognize(
|
|
self,
|
|
message: str,
|
|
available_tools: list[str] | None = None,
|
|
available_skills: list[str] | None = None,
|
|
) -> IntentType:
|
|
"""Recognize user intent.
|
|
|
|
Args:
|
|
message: User message
|
|
available_tools: List of available tool names
|
|
available_skills: List of available skill names
|
|
|
|
Returns:
|
|
Recognized intent type
|
|
"""
|
|
# Simple heuristics for common cases (fast path)
|
|
intent = self._heuristic_recognition(message)
|
|
if intent != IntentType.UNKNOWN:
|
|
logger.info(f"Intent recognized (heuristic): {intent.value} for message: {message[:50]}...")
|
|
return intent
|
|
|
|
# Use LLM for complex cases
|
|
if self._llm_provider:
|
|
return self._llm_recognition(message)
|
|
|
|
# Default to simple if no LLM
|
|
return IntentType.SIMPLE
|
|
|
|
def _heuristic_recognition(self, message: str) -> IntentType:
|
|
"""Fast heuristic-based intent recognition.
|
|
|
|
Args:
|
|
message: User message
|
|
|
|
Returns:
|
|
Recognized intent or UNKNOWN
|
|
"""
|
|
if not message:
|
|
return IntentType.UNKNOWN
|
|
|
|
message_lower = message.lower().strip()
|
|
|
|
# Greetings
|
|
greetings = ["你好", "hello", "hi", "嗨", "您好", "hey"]
|
|
if any(g in message_lower for g in greetings) and len(message_lower) < 20:
|
|
return IntentType.SIMPLE
|
|
|
|
# Simple questions patterns
|
|
simple_patterns = [
|
|
"什么是", "什么叫", "什么是",
|
|
"介绍一下", "请介绍",
|
|
"解释一下", "解释",
|
|
"怎么样", "好不好",
|
|
"是什么意思",
|
|
"who are", "what is", "what's",
|
|
"tell me about",
|
|
]
|
|
|
|
# Check for simple patterns that don't require tools
|
|
for pattern in simple_patterns:
|
|
if pattern in message_lower:
|
|
# But exclude if explicitly asking for current/latest/real-time
|
|
if any(kw in message_lower for kw in ["最新", "现在", "current", "latest", "实时"]):
|
|
return IntentType.UNKNOWN # Might need web search
|
|
return IntentType.SIMPLE
|
|
|
|
# Explicit tool request patterns
|
|
tool_patterns = [
|
|
"搜索", "查找", "search",
|
|
"执行", "运行", "run",
|
|
"计算", "calculate",
|
|
"帮我写代码", "write code",
|
|
"读取", "读取", "read file",
|
|
"创建文件", "write file",
|
|
]
|
|
|
|
for pattern in tool_patterns:
|
|
if pattern in message_lower:
|
|
return IntentType.TOOL
|
|
|
|
# Skill patterns
|
|
skill_patterns = [
|
|
"用python", "用java", "用js",
|
|
"写一个算法", "实现",
|
|
"创建一个", "开发",
|
|
"分析", "优化",
|
|
]
|
|
|
|
for pattern in skill_patterns:
|
|
if pattern in message_lower:
|
|
return IntentType.SKILL
|
|
|
|
# Team patterns
|
|
team_patterns = [
|
|
"团队", "协作", "多个agent",
|
|
"team", "collaborate", "一起",
|
|
]
|
|
|
|
for pattern in team_patterns:
|
|
if pattern in message_lower:
|
|
return IntentType.TEAM
|
|
|
|
return IntentType.UNKNOWN
|
|
|
|
def _llm_recognition(self, message: str) -> IntentType:
|
|
"""LLM-based intent recognition.
|
|
|
|
Args:
|
|
message: User message
|
|
|
|
Returns:
|
|
Recognized intent type
|
|
"""
|
|
try:
|
|
prompt = INTENT_PROMPT.format(message=message)
|
|
|
|
# Use the LLM to classify intent
|
|
response = self._llm_provider.chat(
|
|
messages=[{"role": "user", "content": prompt}],
|
|
max_tokens=50,
|
|
)
|
|
|
|
content = response.content.strip().lower()
|
|
|
|
# Parse the response
|
|
if "simple" in content:
|
|
return IntentType.SIMPLE
|
|
elif "tool" in content:
|
|
return IntentType.TOOL
|
|
elif "skill" in content:
|
|
return IntentType.SKILL
|
|
elif "team" in content:
|
|
return IntentType.TEAM
|
|
else:
|
|
logger.warning(f"Unexpected intent response: {content}")
|
|
return IntentType.SIMPLE # Default to simple
|
|
|
|
except Exception as e:
|
|
logger.error(f"LLM intent recognition failed: {e}")
|
|
return IntentType.SIMPLE # Default to simple on error
|
|
|
|
|
|
class IntentRouter:
|
|
"""Routes requests based on recognized intent."""
|
|
|
|
def __init__(
|
|
self,
|
|
intent_recognizer: IntentRecognizer | None = None,
|
|
use_llm_recognition: bool = True,
|
|
):
|
|
"""Initialize intent router.
|
|
|
|
Args:
|
|
intent_recognizer: Intent recognizer instance
|
|
use_llm_recognition: Whether to use LLM for complex cases
|
|
"""
|
|
self._recognizer = intent_recognizer
|
|
self._use_llm = use_llm_recognition
|
|
|
|
def route(
|
|
self,
|
|
message: str,
|
|
available_tools: list[str] | None = None,
|
|
available_skills: list[str] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Route the user message based on intent.
|
|
|
|
Args:
|
|
message: User message
|
|
available_tools: List of available tool names
|
|
available_skills: List of available skill names
|
|
|
|
Returns:
|
|
Routing decision with intent type and suggested action
|
|
"""
|
|
# Recognize intent
|
|
intent = self._recognizer.recognize(
|
|
message,
|
|
available_tools,
|
|
available_skills,
|
|
)
|
|
|
|
# Build routing decision
|
|
decision = {
|
|
"intent": intent.value,
|
|
"action": self._get_action(intent),
|
|
"message": message,
|
|
}
|
|
|
|
logger.info(f"Routed message to {intent.value}: {message[:50]}...")
|
|
|
|
return decision
|
|
|
|
def _get_action(self, intent: IntentType) -> str:
|
|
"""Get the action to take based on intent.
|
|
|
|
Args:
|
|
intent: Recognized intent type
|
|
|
|
Returns:
|
|
Action name
|
|
"""
|
|
return {
|
|
IntentType.SIMPLE: "direct_response",
|
|
IntentType.TOOL: "execute_tools",
|
|
IntentType.SKILL: "execute_skill",
|
|
IntentType.TEAM: "team_collaboration",
|
|
IntentType.UNKNOWN: "direct_response", # Default to direct response
|
|
}.get(intent, "direct_response")
|
|
|
|
|
|
def create_intent_router(llm_provider=None) -> IntentRouter:
|
|
"""Create an intent router with default settings.
|
|
|
|
Args:
|
|
llm_provider: LLM provider for intent recognition
|
|
|
|
Returns:
|
|
Configured IntentRouter instance
|
|
"""
|
|
recognizer = IntentRecognizer(llm_provider=llm_provider)
|
|
return IntentRouter(intent_recognizer=recognizer)
|