440 lines
16 KiB
Python
440 lines
16 KiB
Python
"""
|
||
意图识别节点
|
||
|
||
使用 LLM 进行智能意图分类，根据用户输入自动路由到对应分支
|
||
支持原生 Function Calling（更准确）和文本解析（回退方案）
|
||
"""
|
||
import json
|
||
import logging
|
||
import time
|
||
from typing import Any, Dict, List
|
||
|
||
from ..base import BaseNode, NodeContext, NodeResult
|
||
from ..registry import NodeRegistry
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
@NodeRegistry.register
|
||
class IntentNode(BaseNode):
|
||
"""
|
||
意图识别节点
|
||
|
||
使用 LLM 分析用户输入，识别用户意图，并路由到对应分支
|
||
支持定义多个意图，每个意图可配置名称、描述和示例
|
||
"""
|
||
|
||
node_type = 'intent'
|
||
node_name = '意图识别'
|
||
node_category = 'logic'
|
||
node_icon = 'brain'
|
||
node_description = '使用 AI 识别用户意图，自动路由到对应分支'
|
||
|
||
supports_branches = True
|
||
|
||
inputs = [
|
||
{
|
||
'name': 'user_input',
|
||
'type': 'string',
|
||
'description': '用户输入文本',
|
||
},
|
||
]
|
||
|
||
outputs = [
|
||
{
|
||
'name': 'intent',
|
||
'type': 'string',
|
||
'description': '识别到的意图名称',
|
||
},
|
||
{
|
||
'name': 'confidence',
|
||
'type': 'number',
|
||
'description': '置信度（0-1）',
|
||
},
|
||
]
|
||
|
||
# 默认的意图识别 prompt 模板
|
||
DEFAULT_SYSTEM_PROMPT = """你是一个意图分类器。根据用户输入，判断用户的意图属于以下哪个类别。
|
||
|
||
可选的意图类别：
|
||
{intents_description}
|
||
|
||
请严格按照以下 JSON 格式输出，不要输出其他任何内容：
|
||
{{"intent": "意图名称", "confidence": 0.95}}
|
||
|
||
注意：
|
||
1. intent 必须是上述意图类别中的一个名称，如果都不匹配则输出 "other"
|
||
2. confidence 是你对这个分类的置信度，范围 0-1
|
||
3. 只输出 JSON，不要有任何解释"""
|
||
|
||
def execute(self, context: NodeContext) -> NodeResult:
|
||
"""执行意图识别（同步方法，通过运行异步方法实现）"""
|
||
import asyncio
|
||
|
||
# 在同步方法中运行异步代码
|
||
loop = asyncio.get_event_loop()
|
||
if loop.is_running():
|
||
# 如果已有事件循环在运行，创建新任务
|
||
import concurrent.futures
|
||
with concurrent.futures.ThreadPoolExecutor() as executor:
|
||
future = executor.submit(asyncio.run, self.execute_async(context))
|
||
return future.result()
|
||
else:
|
||
return loop.run_until_complete(self.execute_async(context))
|
||
|
||
async def execute_async(self, context: NodeContext) -> NodeResult:
|
||
"""异步执行意图识别"""
|
||
start_time = time.time()
|
||
|
||
try:
|
||
from ai_platform.services.llm_service import LLMService
|
||
from ai_platform.models import LLMModel
|
||
|
||
# 获取配置
|
||
model_id = self.config.get('model_id', '')
|
||
intents = self.config.get('intents', [])
|
||
input_variable = self.config.get('input_variable', 'user_input')
|
||
confidence_threshold = self.config.get('confidence_threshold', 0.6)
|
||
|
||
# 获取用户输入
|
||
user_input = context.get_variable(input_variable)
|
||
if not user_input:
|
||
# 尝试从 __user_input__ 获取
|
||
user_input = context.get_variable('__user_input__')
|
||
if not user_input:
|
||
user_input = context.user_input
|
||
|
||
if not user_input:
|
||
return NodeResult(
|
||
success=False,
|
||
error='未获取到用户输入',
|
||
)
|
||
|
||
if not intents:
|
||
return NodeResult(
|
||
success=False,
|
||
error='请至少配置一个意图',
|
||
)
|
||
|
||
llm_service = LLMService(context.db_session)
|
||
model_id = await llm_service.resolve_chat_model_id(model_id)
|
||
|
||
# 检查模型是否支持 Function Calling
|
||
supports_fc = self._check_model_supports_function_call(model_id)
|
||
|
||
if supports_fc:
|
||
# 使用 Function Calling 方式（更准确）
|
||
intent_name, confidence, total_tokens = await self._execute_with_function_calling_async(
|
||
llm_service, model_id, user_input, intents
|
||
)
|
||
else:
|
||
# 回退到文本解析方式
|
||
intent_name, confidence, total_tokens = await self._execute_with_text_parsing_async(
|
||
llm_service, model_id, user_input, intents
|
||
)
|
||
|
||
logger.info(f'意图识别结果: intent_name={intent_name}, confidence={confidence}, use_fc={supports_fc}')
|
||
|
||
elapsed_time = int((time.time() - start_time) * 1000)
|
||
|
||
# 判断是否达到置信度阈值
|
||
if confidence < confidence_threshold:
|
||
logger.info(f'置信度 {confidence} 低于阈值 {confidence_threshold}，走 other 分支')
|
||
intent_name = 'other'
|
||
|
||
# 查找对应的分支 ID
|
||
next_node_id = self._get_branch_id(intent_name, intents)
|
||
logger.info(f'意图识别结果: intent={intent_name}, next_node_id={next_node_id}')
|
||
|
||
# 设置输出变量
|
||
output_var = self.config.get('output_variable', 'intent_result')
|
||
|
||
return NodeResult(
|
||
success=True,
|
||
output=intent_name,
|
||
next_node_id=next_node_id,
|
||
output_variables={
|
||
output_var: intent_name,
|
||
f'{output_var}_confidence': confidence,
|
||
f'{output_var}_input': user_input,
|
||
},
|
||
tokens_used=total_tokens,
|
||
elapsed_time=elapsed_time,
|
||
metadata={
|
||
'intent': intent_name,
|
||
'confidence': confidence,
|
||
'matched_branch': next_node_id,
|
||
'model_id': model_id,
|
||
},
|
||
)
|
||
|
||
except Exception as e:
|
||
logger.exception(f'意图识别节点执行失败: {e}')
|
||
elapsed_time = int((time.time() - start_time) * 1000)
|
||
return NodeResult(
|
||
success=False,
|
||
error=str(e),
|
||
elapsed_time=elapsed_time,
|
||
)
|
||
|
||
def _check_model_supports_function_call(self, model_id) -> bool:
|
||
"""检查模型是否支持 Function Calling
|
||
|
||
TODO: 此方法需要改造为异步版本，当前使用同步查询作为临时方案
|
||
"""
|
||
# 临时方案：默认返回True，让调用方尝试使用Function Calling
|
||
# 实际使用时应通过上下文传入模型信息
|
||
return True
|
||
|
||
async def _execute_with_function_calling_async(
|
||
self,
|
||
llm_service,
|
||
model_id: str,
|
||
user_input: str,
|
||
intents: List[Dict],
|
||
) -> tuple:
|
||
"""使用 Function Calling 执行意图识别（更准确，异步版本）"""
|
||
# 构建意图名称列表（用于 enum）
|
||
intent_names = [intent.get('name') for intent in intents] + ['other']
|
||
|
||
# 构建意图描述（用于 LLM 理解）
|
||
intents_description = self._build_intents_description(intents)
|
||
|
||
# 构建 Function Calling 工具定义
|
||
tools = [{
|
||
'name': 'classify_intent',
|
||
'description': f'根据用户输入对意图进行分类。可选的意图类别：\n{intents_description}',
|
||
'parameters': {
|
||
'type': 'object',
|
||
'properties': {
|
||
'intent': {
|
||
'type': 'string',
|
||
'enum': intent_names,
|
||
'description': '识别到的意图名称',
|
||
},
|
||
'confidence': {
|
||
'type': 'number',
|
||
'description': '置信度，范围 0-1',
|
||
},
|
||
},
|
||
'required': ['intent', 'confidence'],
|
||
},
|
||
}]
|
||
|
||
# 简化的系统提示词
|
||
system_prompt = "你是一个意图分类器。分析用户输入，调用 classify_intent 函数返回分类结果。"
|
||
|
||
messages = [
|
||
{'role': 'system', 'content': system_prompt},
|
||
{'role': 'user', 'content': user_input},
|
||
]
|
||
|
||
# 调用 LLM（带 Function Calling，异步）
|
||
response = await llm_service.chat_async(
|
||
model_id=model_id,
|
||
messages=messages,
|
||
temperature=0.1,
|
||
max_tokens=100,
|
||
tools=tools,
|
||
tool_choice='required', # 强制使用 Function Calling
|
||
)
|
||
|
||
# 从 tool_calls 中提取结果
|
||
if response.has_tool_calls and response.tool_calls:
|
||
tool_call = response.tool_calls[0]
|
||
args = tool_call.arguments
|
||
intent_name = args.get('intent', 'other')
|
||
confidence = float(args.get('confidence', 0.8))
|
||
|
||
# 验证意图名称
|
||
if intent_name not in intent_names:
|
||
intent_name = 'other'
|
||
confidence = 0.5
|
||
|
||
return intent_name, confidence, response.total_tokens
|
||
else:
|
||
# Function Calling 失败，返回默认值
|
||
logger.warning('Function Calling 未返回 tool_calls，回退到 other')
|
||
return 'other', 0.5, response.total_tokens
|
||
|
||
async def _execute_with_text_parsing_async(
|
||
self,
|
||
llm_service,
|
||
model_id: str,
|
||
user_input: str,
|
||
intents: List[Dict],
|
||
) -> tuple:
|
||
"""使用文本解析执行意图识别（回退方案，异步版本）"""
|
||
# 构建意图描述
|
||
intents_description = self._build_intents_description(intents)
|
||
|
||
# 构建 prompt
|
||
system_prompt = self.DEFAULT_SYSTEM_PROMPT.format(
|
||
intents_description=intents_description
|
||
)
|
||
|
||
messages = [
|
||
{'role': 'system', 'content': system_prompt},
|
||
{'role': 'user', 'content': user_input},
|
||
]
|
||
|
||
# 调用 LLM（异步）
|
||
response = await llm_service.chat_async(
|
||
model_id=model_id,
|
||
messages=messages,
|
||
temperature=0.1,
|
||
max_tokens=100,
|
||
)
|
||
|
||
# 解析响应
|
||
intent_name, confidence = self._parse_response(response.content, intents)
|
||
|
||
return intent_name, confidence, response.total_tokens
|
||
|
||
def _build_intents_description(self, intents: List[Dict]) -> str:
|
||
"""构建意图描述文本"""
|
||
lines = []
|
||
for i, intent in enumerate(intents, 1):
|
||
name = intent.get('name', '')
|
||
description = intent.get('description', '')
|
||
examples = intent.get('examples', [])
|
||
|
||
line = f"{i}. {name}"
|
||
if description:
|
||
line += f" - {description}"
|
||
|
||
lines.append(line)
|
||
|
||
# 添加示例
|
||
if examples:
|
||
for example in examples[:3]: # 最多3个示例
|
||
lines.append(f" 示例: \"{example}\"")
|
||
|
||
# 添加 other 选项
|
||
lines.append(f"{len(intents) + 1}. other - 以上都不匹配时选择此项")
|
||
|
||
return '\n'.join(lines)
|
||
|
||
def _parse_response(self, response: str, intents: List[Dict]) -> tuple:
|
||
"""解析 LLM 响应"""
|
||
logger.info(f'开始解析 LLM 响应: {response}')
|
||
try:
|
||
# 尝试提取 JSON
|
||
response = response.strip()
|
||
|
||
# 处理可能的 markdown 代码块
|
||
if response.startswith('```'):
|
||
lines = response.split('\n')
|
||
json_lines = []
|
||
in_json = False
|
||
for line in lines:
|
||
if line.startswith('```') and not in_json:
|
||
in_json = True
|
||
continue
|
||
elif line.startswith('```') and in_json:
|
||
break
|
||
elif in_json:
|
||
json_lines.append(line)
|
||
response = '\n'.join(json_lines)
|
||
|
||
# 解析 JSON
|
||
result = json.loads(response)
|
||
intent_name = result.get('intent', 'other')
|
||
confidence = float(result.get('confidence', 0.5))
|
||
|
||
# 验证意图名称是否有效
|
||
valid_names = [intent.get('name') for intent in intents] + ['other']
|
||
if intent_name not in valid_names:
|
||
# 尝试模糊匹配
|
||
intent_name_lower = intent_name.lower()
|
||
for valid_name in valid_names:
|
||
if valid_name.lower() == intent_name_lower:
|
||
intent_name = valid_name
|
||
break
|
||
else:
|
||
intent_name = 'other'
|
||
confidence = 0.5
|
||
|
||
return intent_name, confidence
|
||
|
||
except (json.JSONDecodeError, ValueError, KeyError) as e:
|
||
logger.warning(f'解析意图识别响应失败: {e}, response: {response}')
|
||
return 'other', 0.5
|
||
|
||
def _get_branch_id(self, intent_name: str, intents: List[Dict]) -> str:
|
||
"""获取意图对应的分支 ID"""
|
||
logger.info(f'查找意图分支: intent_name={intent_name}, intents={intents}')
|
||
for intent in intents:
|
||
if intent.get('name') == intent_name:
|
||
branch_id = intent.get('branch_id', intent_name)
|
||
logger.info(f'找到匹配意图: {intent}, 返回 branch_id={branch_id}')
|
||
return branch_id
|
||
logger.info(f'未找到匹配意图，返回 other')
|
||
return 'other'
|
||
|
||
@classmethod
|
||
def get_config_schema(cls) -> Dict[str, Any]:
|
||
"""获取配置 Schema"""
|
||
return {
|
||
'type': 'object',
|
||
'properties': {
|
||
'model_id': {
|
||
'type': 'string',
|
||
'title': '模型',
|
||
'description': '选择用于意图识别的 LLM 模型',
|
||
},
|
||
'input_variable': {
|
||
'type': 'string',
|
||
'title': '输入变量',
|
||
'description': '包含用户输入的变量名',
|
||
'default': 'user_input',
|
||
},
|
||
'intents': {
|
||
'type': 'array',
|
||
'title': '意图列表',
|
||
'description': '定义要识别的意图',
|
||
'items': {
|
||
'type': 'object',
|
||
'properties': {
|
||
'name': {
|
||
'type': 'string',
|
||
'title': '意图名称',
|
||
'description': '唯一标识，如 consult, complaint',
|
||
},
|
||
'description': {
|
||
'type': 'string',
|
||
'title': '意图描述',
|
||
'description': '描述这个意图的含义',
|
||
},
|
||
'examples': {
|
||
'type': 'array',
|
||
'title': '示例',
|
||
'description': '用户可能的输入示例',
|
||
'items': {'type': 'string'},
|
||
},
|
||
'branch_id': {
|
||
'type': 'string',
|
||
'title': '分支 ID',
|
||
'description': '匹配时跳转的分支',
|
||
},
|
||
},
|
||
'required': ['name'],
|
||
},
|
||
},
|
||
'confidence_threshold': {
|
||
'type': 'number',
|
||
'title': '置信度阈值',
|
||
'description': '低于此阈值将走 other 分支',
|
||
'default': 0.6,
|
||
'minimum': 0,
|
||
'maximum': 1,
|
||
},
|
||
'output_variable': {
|
||
'type': 'string',
|
||
'title': '输出变量名',
|
||
'default': 'intent_result',
|
||
},
|
||
},
|
||
'required': ['model_id', 'intents'],
|
||
}
|