1070 lines
44 KiB
Python
1070 lines
44 KiB
Python
"""
|
||
LLM 调用节点
|
||
|
||
支持普通文本输出和结构化输出（基于 Function Calling）
|
||
"""
|
||
import json
|
||
import logging
|
||
import time
|
||
from dataclasses import dataclass
|
||
from typing import Any, Dict, Generator, List
|
||
|
||
from ..base import BaseNode, NodeContext, NodeResult
|
||
from ..registry import NodeRegistry
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
@dataclass
|
||
class LLMStreamChunkEvent:
|
||
"""LLM 流式输出事件"""
|
||
content: str = '' # 增量内容
|
||
is_finished: bool = False # 是否完成
|
||
accumulated_content: str = '' # 累积内容
|
||
tokens_used: int = 0 # Token 使用量（完成时才有）
|
||
|
||
|
||
@NodeRegistry.register
|
||
class LLMNode(BaseNode):
|
||
"""
|
||
LLM 调用节点
|
||
|
||
调用大语言模型进行对话或生成
|
||
"""
|
||
|
||
node_type = 'llm'
|
||
node_name = 'LLM'
|
||
node_category = 'llm'
|
||
node_icon = 'bot'
|
||
node_description = '调用大语言模型进行对话或生成'
|
||
|
||
inputs = [
|
||
{
|
||
'name': 'prompt',
|
||
'type': 'string',
|
||
'description': '提示词',
|
||
},
|
||
{
|
||
'name': 'context',
|
||
'type': 'string',
|
||
'description': '上下文（可选）',
|
||
},
|
||
]
|
||
|
||
outputs = [
|
||
{
|
||
'name': 'response',
|
||
'type': 'string',
|
||
'description': 'LLM 响应',
|
||
},
|
||
{
|
||
'name': 'tokens',
|
||
'type': 'number',
|
||
'description': '使用的 Token 数',
|
||
},
|
||
]
|
||
|
||
def execute(self, context: NodeContext) -> NodeResult:
|
||
"""执行 LLM 调用（同步方法，通过运行异步方法实现）"""
|
||
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:
|
||
"""异步执行 LLM 调用"""
|
||
start_time = time.time()
|
||
|
||
try:
|
||
from ai_platform.services.llm_service import LLMService
|
||
|
||
# 获取配置
|
||
model_id = self.config.get('model_id', '')
|
||
system_prompt = self.config.get('system_prompt', '')
|
||
user_prompt = self.config.get('user_prompt', '')
|
||
temperature = self.config.get('temperature', 0.7)
|
||
max_tokens = self.config.get('max_tokens', 2048)
|
||
|
||
# 结构化输出配置
|
||
output_mode = self.config.get('output_mode', 'text')
|
||
output_schema_source = self.config.get('output_schema_source', 'static')
|
||
output_schema = self.config.get('output_schema', [])
|
||
|
||
# 如果结构来源是变量，从上下文中获取
|
||
if output_schema_source == 'variable':
|
||
output_schema_variable = self.config.get('output_schema_variable', '')
|
||
if output_schema_variable:
|
||
# 直接从上下文变量中获取原始值（不使用 resolve_template，因为它会将对象转为字符串）
|
||
# 支持格式：{{node_id.variable_name}} 或 variable_name
|
||
# 从 {{node.var}} 中提取最后一个部分作为变量名
|
||
var_name = output_schema_variable.strip()
|
||
if var_name.startswith('{{') and var_name.endswith('}}'):
|
||
var_name = var_name[2:-2].strip()
|
||
|
||
# 如果是 node.variable 格式，提取 variable 部分
|
||
if '.' in var_name:
|
||
var_name = var_name.split('.')[-1]
|
||
|
||
# 尝试从上下文变量中获取
|
||
resolved = context.variables.get(var_name)
|
||
|
||
if resolved is not None:
|
||
try:
|
||
import json
|
||
# 优先检查是否已经是 list 类型（直接从上下文变量获取）
|
||
if isinstance(resolved, list):
|
||
output_schema = resolved
|
||
elif isinstance(resolved, str):
|
||
# 尝试解析 JSON 字符串
|
||
output_schema = json.loads(resolved)
|
||
logger.info(f'LLM 从变量 {var_name} 获取结构化输出配置: {len(output_schema)} 个字段')
|
||
except (json.JSONDecodeError, TypeError) as e:
|
||
logger.warning(f'解析结构化输出变量失败: {e}')
|
||
logger.warning(f'var_name: {var_name}, resolved 类型: {type(resolved)}, 值: {str(resolved)[:200]}')
|
||
|
||
logger.info(f'LLM execute_async - 原始 user_prompt: {user_prompt[:200] if user_prompt else "无"}')
|
||
logger.info(f'LLM execute_async - 上下文变量: {context.variables}')
|
||
|
||
# 解析模板变量
|
||
if system_prompt:
|
||
system_prompt = context.resolve_template(system_prompt)
|
||
if user_prompt:
|
||
user_prompt = context.resolve_template(user_prompt)
|
||
|
||
logger.info(f'LLM execute_async - 解析后 user_prompt: {user_prompt[:200] if user_prompt else "无"}')
|
||
|
||
# 至少需要配置 system_prompt 或 user_prompt 之一
|
||
if not system_prompt and not user_prompt:
|
||
return NodeResult(
|
||
success=False,
|
||
error='请至少配置系统提示词或用户提示词',
|
||
)
|
||
|
||
# 构建消息
|
||
messages = []
|
||
if system_prompt:
|
||
messages.append({'role': 'system', 'content': system_prompt})
|
||
|
||
# 添加对话历史
|
||
if self.config.get('use_conversation_history', False):
|
||
messages.extend(context.conversation_history)
|
||
|
||
# 只有配置了 user_prompt 才添加用户消息
|
||
if user_prompt:
|
||
messages.append({'role': 'user', 'content': user_prompt})
|
||
|
||
# 调用 LLM（异步）
|
||
llm_service = LLMService(context.db_session)
|
||
model_id = await llm_service.resolve_chat_model_id(model_id)
|
||
output_var = self.config.get('output_variable', 'llm_response')
|
||
|
||
# 根据输出模式选择执行方式
|
||
if output_mode == 'structured' and output_schema:
|
||
# 结构化输出模式：使用 Function Calling（异步）
|
||
result = await self._execute_structured_output_async(
|
||
llm_service, model_id, messages, temperature, max_tokens,
|
||
output_schema, output_var, start_time
|
||
)
|
||
result.metadata = {
|
||
**(result.metadata or {}),
|
||
'model_id': str(model_id),
|
||
'output_variable': output_var,
|
||
}
|
||
return result
|
||
else:
|
||
# 普通文本输出模式（异步）
|
||
response = await llm_service.chat_async(
|
||
model_id=model_id,
|
||
messages=messages,
|
||
temperature=temperature,
|
||
max_tokens=max_tokens,
|
||
)
|
||
|
||
elapsed_time = int((time.time() - start_time) * 1000)
|
||
|
||
return NodeResult(
|
||
success=True,
|
||
output=response.content,
|
||
output_variables={
|
||
output_var: response.content,
|
||
f'{output_var}_tokens': response.total_tokens,
|
||
},
|
||
tokens_used=response.total_tokens,
|
||
elapsed_time=elapsed_time,
|
||
metadata={
|
||
'model_id': str(model_id),
|
||
'model': response.model,
|
||
'prompt_tokens': response.prompt_tokens,
|
||
'completion_tokens': response.completion_tokens,
|
||
'output_variable': output_var,
|
||
},
|
||
)
|
||
|
||
except Exception as e:
|
||
logger.exception(f'LLM 节点异步执行失败: {e}')
|
||
elapsed_time = int((time.time() - start_time) * 1000)
|
||
return NodeResult(
|
||
success=False,
|
||
error=str(e),
|
||
elapsed_time=elapsed_time,
|
||
)
|
||
|
||
def execute_stream(self, context: NodeContext) -> Generator[LLMStreamChunkEvent, None, NodeResult]:
|
||
"""
|
||
流式执行 LLM 调用
|
||
|
||
Yields:
|
||
LLMStreamChunkEvent: 流式输出事件
|
||
|
||
Returns:
|
||
NodeResult: 最终执行结果
|
||
"""
|
||
start_time = time.time()
|
||
|
||
try:
|
||
from ai_platform.services.llm_service import LLMService
|
||
|
||
# 获取配置
|
||
model_id = self.config.get('model_id', '')
|
||
system_prompt = self.config.get('system_prompt', '')
|
||
user_prompt = self.config.get('user_prompt', '')
|
||
temperature = self.config.get('temperature', 0.7)
|
||
max_tokens = self.config.get('max_tokens', 2048)
|
||
|
||
# 结构化输出配置
|
||
output_mode = self.config.get('output_mode', 'text') # text 或 structured
|
||
output_schema_source = self.config.get('output_schema_source', 'static')
|
||
output_schema = self.config.get('output_schema', [])
|
||
|
||
# 如果结构来源是变量，从上下文中获取
|
||
if output_schema_source == 'variable':
|
||
output_schema_variable = self.config.get('output_schema_variable', '')
|
||
if output_schema_variable:
|
||
# 直接从上下文变量中获取原始值（不使用 resolve_template，因为它会将对象转为字符串）
|
||
# 支持格式：{{node_id.variable_name}} 或 variable_name
|
||
# 从 {{node.var}} 中提取最后一个部分作为变量名
|
||
var_name = output_schema_variable.strip()
|
||
if var_name.startswith('{{') and var_name.endswith('}}'):
|
||
var_name = var_name[2:-2].strip()
|
||
|
||
# 如果是 node.variable 格式，提取 variable 部分
|
||
if '.' in var_name:
|
||
var_name = var_name.split('.')[-1]
|
||
|
||
# 尝试从上下文变量中获取
|
||
resolved = context.variables.get(var_name)
|
||
|
||
if resolved is not None:
|
||
try:
|
||
import json
|
||
# 优先检查是否已经是 list 类型（直接从上下文变量获取）
|
||
if isinstance(resolved, list):
|
||
output_schema = resolved
|
||
elif isinstance(resolved, str):
|
||
# 尝试解析 JSON 字符串
|
||
output_schema = json.loads(resolved)
|
||
logger.info(f'LLM stream 从变量 {var_name} 获取结构化输出配置: {len(output_schema)} 个字段')
|
||
except (json.JSONDecodeError, TypeError) as e:
|
||
logger.warning(f'解析结构化输出变量失败: {e}')
|
||
logger.warning(f'var_name: {var_name}, resolved 类型: {type(resolved)}, 值: {str(resolved)[:200]}')
|
||
|
||
logger.info(f'LLM stream - 原始 user_prompt: {user_prompt[:200] if user_prompt else "无"}')
|
||
logger.info(f'LLM stream - 上下文变量: {context.variables}')
|
||
|
||
# 解析模板变量
|
||
if system_prompt:
|
||
system_prompt = context.resolve_template(system_prompt)
|
||
if user_prompt:
|
||
user_prompt = context.resolve_template(user_prompt)
|
||
|
||
logger.info(f'LLM stream - 解析后 user_prompt: {user_prompt[:200] if user_prompt else "无"}')
|
||
|
||
# 至少需要配置 system_prompt 或 user_prompt 之一
|
||
if not system_prompt and not user_prompt:
|
||
yield LLMStreamChunkEvent(
|
||
content='',
|
||
is_finished=True,
|
||
accumulated_content='',
|
||
)
|
||
return NodeResult(
|
||
success=False,
|
||
error='请至少配置系统提示词或用户提示词',
|
||
)
|
||
|
||
# 构建消息
|
||
messages = []
|
||
if system_prompt:
|
||
messages.append({'role': 'system', 'content': system_prompt})
|
||
|
||
# 添加对话历史
|
||
if self.config.get('use_conversation_history', False):
|
||
messages.extend(context.conversation_history)
|
||
|
||
# 只有配置了 user_prompt 才添加用户消息
|
||
if user_prompt:
|
||
messages.append({'role': 'user', 'content': user_prompt})
|
||
|
||
llm_service = LLMService(context.db_session)
|
||
output_var = self.config.get('output_variable', 'llm_response')
|
||
|
||
# 根据输出模式选择执行方式
|
||
if output_mode == 'structured' and output_schema:
|
||
# 结构化输出模式：使用 Function Calling 流式调用
|
||
result = yield from self._execute_stream_structured_output(
|
||
llm_service, model_id, messages, temperature, max_tokens,
|
||
output_schema, output_var, start_time
|
||
)
|
||
result.metadata = {
|
||
**(result.metadata or {}),
|
||
'model_id': str(model_id),
|
||
'output_variable': output_var,
|
||
}
|
||
return result
|
||
else:
|
||
# 普通文本输出模式
|
||
accumulated_content = ''
|
||
total_tokens = 0
|
||
|
||
for chunk in llm_service.chat_stream_sync(
|
||
model_id=model_id,
|
||
messages=messages,
|
||
temperature=temperature,
|
||
max_tokens=max_tokens,
|
||
):
|
||
accumulated_content += chunk.content
|
||
|
||
# 如果是最后一个 chunk，获取 token 信息
|
||
if chunk.is_finished and chunk.total_tokens > 0:
|
||
total_tokens = chunk.total_tokens
|
||
|
||
yield LLMStreamChunkEvent(
|
||
content=chunk.content,
|
||
is_finished=chunk.is_finished,
|
||
accumulated_content=accumulated_content,
|
||
tokens_used=total_tokens,
|
||
)
|
||
|
||
elapsed_time = int((time.time() - start_time) * 1000)
|
||
|
||
return NodeResult(
|
||
success=True,
|
||
output=accumulated_content,
|
||
output_variables={
|
||
output_var: accumulated_content,
|
||
f'{output_var}_tokens': total_tokens,
|
||
},
|
||
tokens_used=total_tokens,
|
||
elapsed_time=elapsed_time,
|
||
metadata={
|
||
'model_id': str(model_id),
|
||
'output_variable': output_var,
|
||
},
|
||
)
|
||
|
||
except Exception as e:
|
||
logger.exception(f'LLM 节点流式执行失败: {e}')
|
||
elapsed_time = int((time.time() - start_time) * 1000)
|
||
yield LLMStreamChunkEvent(
|
||
content='',
|
||
is_finished=True,
|
||
accumulated_content='',
|
||
)
|
||
return NodeResult(
|
||
success=False,
|
||
error=str(e),
|
||
elapsed_time=elapsed_time,
|
||
)
|
||
|
||
def _execute_stream_structured_output(
|
||
self,
|
||
llm_service,
|
||
model_id: str,
|
||
messages: List[Dict],
|
||
temperature: float,
|
||
max_tokens: int,
|
||
output_schema: List[Dict],
|
||
output_var: str,
|
||
start_time: float,
|
||
) -> Generator[LLMStreamChunkEvent, None, NodeResult]:
|
||
"""
|
||
使用 Function Calling 流式执行结构化输出
|
||
|
||
Args:
|
||
llm_service: LLM 服务实例
|
||
model_id: 模型 ID
|
||
messages: 消息列表
|
||
temperature: 温度
|
||
max_tokens: 最大 token
|
||
output_schema: 输出字段定义列表（支持多层嵌套）
|
||
output_var: 输出变量名
|
||
start_time: 开始时间
|
||
|
||
Yields:
|
||
LLMStreamChunkEvent: 流式输出事件
|
||
|
||
Returns:
|
||
NodeResult: 执行结果
|
||
"""
|
||
# 使用递归方法构建多层嵌套的 JSON Schema
|
||
properties, required_fields = self._build_schema_properties(output_schema)
|
||
|
||
if not properties:
|
||
yield LLMStreamChunkEvent(
|
||
content='',
|
||
is_finished=True,
|
||
accumulated_content='',
|
||
)
|
||
return NodeResult(
|
||
success=False,
|
||
error='结构化输出模式需要至少定义一个输出字段',
|
||
elapsed_time=int((time.time() - start_time) * 1000),
|
||
)
|
||
|
||
# 构建 Function Calling 工具定义
|
||
tools = [{
|
||
'name': 'extract_structured_data',
|
||
'description': '从文本中提取结构化数据并返回 JSON 格式',
|
||
'parameters': {
|
||
'type': 'object',
|
||
'properties': properties,
|
||
'required': required_fields,
|
||
},
|
||
}]
|
||
|
||
# 检查模型是否支持 Function Calling
|
||
# TODO: 需要改造为异步版本，当前默认支持
|
||
supports_fc = True
|
||
|
||
accumulated_content = ''
|
||
total_tokens = 0
|
||
tool_call_arguments = ''
|
||
|
||
extracted_data = None
|
||
|
||
if supports_fc:
|
||
# 使用 Function Calling 流式调用
|
||
for chunk in llm_service.chat_stream_sync(
|
||
model_id=model_id,
|
||
messages=messages,
|
||
temperature=temperature,
|
||
max_tokens=max_tokens,
|
||
tools=tools,
|
||
tool_choice='required',
|
||
):
|
||
# 处理 tool_calls 流式数据
|
||
if chunk.tool_calls:
|
||
for tc in chunk.tool_calls:
|
||
if tc.arguments:
|
||
# arguments 可能是 dict（已解析）或 str（流式增量）
|
||
if isinstance(tc.arguments, dict):
|
||
# 已解析的完整参数，直接使用
|
||
extracted_data = tc.arguments
|
||
accumulated_content = json.dumps(tc.arguments, ensure_ascii=False)
|
||
else:
|
||
# 字符串形式的增量参数
|
||
tool_call_arguments += str(tc.arguments)
|
||
accumulated_content = tool_call_arguments
|
||
elif chunk.tool_call_delta:
|
||
# 处理增量的 tool_call 数据
|
||
if chunk.tool_call_delta.get('arguments'):
|
||
args = chunk.tool_call_delta['arguments']
|
||
if isinstance(args, str):
|
||
tool_call_arguments += args
|
||
else:
|
||
tool_call_arguments += json.dumps(args, ensure_ascii=False)
|
||
accumulated_content = tool_call_arguments
|
||
elif chunk.content:
|
||
accumulated_content += chunk.content
|
||
|
||
if chunk.is_finished and chunk.total_tokens > 0:
|
||
total_tokens = chunk.total_tokens
|
||
|
||
yield LLMStreamChunkEvent(
|
||
content=chunk.content or '',
|
||
is_finished=chunk.is_finished,
|
||
accumulated_content=accumulated_content,
|
||
tokens_used=total_tokens,
|
||
)
|
||
|
||
elapsed_time = int((time.time() - start_time) * 1000)
|
||
|
||
# 解析 tool_call 参数
|
||
if extracted_data:
|
||
# 已经有解析好的数据（dict 类型）
|
||
json_output = json.dumps(extracted_data, ensure_ascii=False, indent=2)
|
||
|
||
output_variables = {
|
||
output_var: json_output,
|
||
f'{output_var}_data': extracted_data,
|
||
f'{output_var}_tokens': total_tokens,
|
||
}
|
||
|
||
for key, value in extracted_data.items():
|
||
output_variables[f'{output_var}_{key}'] = value
|
||
|
||
logger.info(f'LLM 结构化输出成功 (Function Calling 流式), tokens: {total_tokens}')
|
||
return NodeResult(
|
||
success=True,
|
||
output=json_output,
|
||
output_variables=output_variables,
|
||
tokens_used=total_tokens,
|
||
elapsed_time=elapsed_time,
|
||
metadata={
|
||
'output_mode': 'structured',
|
||
'use_function_calling': True,
|
||
'call_method': 'Function Calling Stream (tool_choice=required)',
|
||
},
|
||
)
|
||
elif tool_call_arguments:
|
||
# 从字符串解析 JSON
|
||
try:
|
||
extracted_data = json.loads(tool_call_arguments)
|
||
json_output = json.dumps(extracted_data, ensure_ascii=False, indent=2)
|
||
|
||
output_variables = {
|
||
output_var: json_output,
|
||
f'{output_var}_data': extracted_data,
|
||
f'{output_var}_tokens': total_tokens,
|
||
}
|
||
|
||
for key, value in extracted_data.items():
|
||
output_variables[f'{output_var}_{key}'] = value
|
||
|
||
logger.info(f'LLM 结构化输出成功 (Function Calling 流式解析), tokens: {total_tokens}')
|
||
return NodeResult(
|
||
success=True,
|
||
output=json_output,
|
||
output_variables=output_variables,
|
||
tokens_used=total_tokens,
|
||
elapsed_time=elapsed_time,
|
||
metadata={
|
||
'output_mode': 'structured',
|
||
'use_function_calling': True,
|
||
'call_method': 'Function Calling Stream (parsed from string)',
|
||
},
|
||
)
|
||
except json.JSONDecodeError as e:
|
||
logger.warning(f'Function Calling 参数解析失败: {e}')
|
||
return self._parse_json_from_text(
|
||
accumulated_content, output_var, total_tokens, elapsed_time
|
||
)
|
||
else:
|
||
# 没有 tool_call 结果，尝试从文本解析
|
||
return self._parse_json_from_text(
|
||
accumulated_content, output_var, total_tokens, elapsed_time
|
||
)
|
||
else:
|
||
# 模型不支持 Function Calling，使用普通流式调用后解析文本
|
||
for chunk in llm_service.chat_stream_sync(
|
||
model_id=model_id,
|
||
messages=messages,
|
||
temperature=temperature,
|
||
max_tokens=max_tokens,
|
||
):
|
||
accumulated_content += chunk.content
|
||
|
||
if chunk.is_finished and chunk.total_tokens > 0:
|
||
total_tokens = chunk.total_tokens
|
||
|
||
yield LLMStreamChunkEvent(
|
||
content=chunk.content,
|
||
is_finished=chunk.is_finished,
|
||
accumulated_content=accumulated_content,
|
||
tokens_used=total_tokens,
|
||
)
|
||
|
||
elapsed_time = int((time.time() - start_time) * 1000)
|
||
|
||
return self._parse_json_from_text(
|
||
accumulated_content, output_var, total_tokens, elapsed_time
|
||
)
|
||
|
||
def _execute_structured_output(
|
||
self,
|
||
llm_service,
|
||
model_id: str,
|
||
messages: List[Dict],
|
||
temperature: float,
|
||
max_tokens: int,
|
||
output_schema: List[Dict],
|
||
output_var: str,
|
||
start_time: float,
|
||
) -> NodeResult:
|
||
"""
|
||
使用 Function Calling 执行结构化输出
|
||
|
||
Args:
|
||
llm_service: LLM 服务实例
|
||
model_id: 模型 ID
|
||
messages: 消息列表
|
||
temperature: 温度
|
||
max_tokens: 最大 token
|
||
output_schema: 输出字段定义列表（支持多层嵌套）
|
||
output_var: 输出变量名
|
||
start_time: 开始时间
|
||
|
||
Returns:
|
||
NodeResult: 执行结果
|
||
"""
|
||
# 使用递归方法构建多层嵌套的 JSON Schema
|
||
properties, required_fields = self._build_schema_properties(output_schema)
|
||
|
||
if not properties:
|
||
return NodeResult(
|
||
success=False,
|
||
error='结构化输出模式需要至少定义一个输出字段',
|
||
elapsed_time=int((time.time() - start_time) * 1000),
|
||
)
|
||
|
||
# 构建 Function Calling 工具定义
|
||
tools = [{
|
||
'name': 'extract_structured_data',
|
||
'description': '从文本中提取结构化数据并返回 JSON 格式',
|
||
'parameters': {
|
||
'type': 'object',
|
||
'properties': properties,
|
||
'required': required_fields,
|
||
},
|
||
}]
|
||
|
||
# 检查模型是否支持 Function Calling
|
||
# TODO: 需要改造为异步版本，当前默认支持
|
||
supports_fc = True
|
||
|
||
if supports_fc:
|
||
# 使用 Function Calling
|
||
response = llm_service.chat(
|
||
model_id=model_id,
|
||
messages=messages,
|
||
temperature=temperature,
|
||
max_tokens=max_tokens,
|
||
tools=tools,
|
||
tool_choice='required',
|
||
)
|
||
|
||
elapsed_time = int((time.time() - start_time) * 1000)
|
||
|
||
if response.has_tool_calls and response.tool_calls:
|
||
# 从 tool_calls 中提取结构化数据
|
||
extracted_data = response.tool_calls[0].arguments
|
||
|
||
# 处理双重序列化问题：如果 extracted_data 是 {"raw": "..."} 格式，提取 raw 字段
|
||
if isinstance(extracted_data, dict) and len(extracted_data) == 1 and 'raw' in extracted_data:
|
||
raw_value = extracted_data['raw']
|
||
logger.info(f'检测到 raw 包装格式，提取 raw 字段: {raw_value[:200] if isinstance(raw_value, str) else raw_value}')
|
||
if isinstance(raw_value, str):
|
||
try:
|
||
extracted_data = json.loads(raw_value)
|
||
logger.info(f'成功解析 raw 字段为 JSON 对象')
|
||
except json.JSONDecodeError as e:
|
||
logger.warning(f'raw 字段不是有效的 JSON: {e}')
|
||
# 保持原样
|
||
else:
|
||
extracted_data = raw_value
|
||
|
||
json_output = json.dumps(extracted_data, ensure_ascii=False, indent=2)
|
||
logger.info(f'LLM 结构化输出成功 (Function Calling 同步), tokens: {response.total_tokens}')
|
||
|
||
# 构建输出变量：既有 JSON 字符串，也有各个字段
|
||
output_variables = {
|
||
output_var: json_output,
|
||
f'{output_var}_data': extracted_data,
|
||
f'{output_var}_tokens': response.total_tokens,
|
||
f'{output_var}_raw': response.tool_calls[0].arguments, # 保留原始响应
|
||
}
|
||
|
||
# 将每个字段也作为独立变量输出
|
||
for key, value in extracted_data.items():
|
||
output_variables[f'{output_var}_{key}'] = value
|
||
|
||
return NodeResult(
|
||
success=True,
|
||
output=json_output,
|
||
output_variables=output_variables,
|
||
tokens_used=response.total_tokens,
|
||
elapsed_time=elapsed_time,
|
||
metadata={
|
||
'model': response.model,
|
||
'output_mode': 'structured',
|
||
'use_function_calling': True,
|
||
'call_method': 'Function Calling (tool_choice=required)',
|
||
},
|
||
)
|
||
else:
|
||
# Function Calling 未返回结果，回退到文本解析
|
||
logger.warning('Function Calling 未返回 tool_calls，尝试解析文本响应')
|
||
return self._parse_json_from_text(
|
||
response.content, output_var, response.total_tokens, elapsed_time
|
||
)
|
||
else:
|
||
# 模型不支持 Function Calling，使用文本解析
|
||
response = llm_service.chat(
|
||
model_id=model_id,
|
||
messages=messages,
|
||
temperature=temperature,
|
||
max_tokens=max_tokens,
|
||
)
|
||
|
||
elapsed_time = int((time.time() - start_time) * 1000)
|
||
|
||
return self._parse_json_from_text(
|
||
response.content, output_var, response.total_tokens, elapsed_time
|
||
)
|
||
|
||
def _build_schema_properties(self, output_schema: List[Dict]) -> tuple:
|
||
"""
|
||
递归构建多层嵌套的 JSON Schema properties
|
||
|
||
Args:
|
||
output_schema: 输出字段定义列表，支持嵌套结构
|
||
|
||
Returns:
|
||
tuple: (properties dict, required fields list)
|
||
"""
|
||
properties = {}
|
||
required_fields = []
|
||
|
||
type_mapping = {
|
||
'string': 'string',
|
||
'number': 'number',
|
||
'integer': 'integer',
|
||
'boolean': 'boolean',
|
||
'array': 'array',
|
||
'object': 'object',
|
||
}
|
||
|
||
for field in output_schema:
|
||
field_name = field.get('name', '')
|
||
field_type = field.get('type', 'string')
|
||
field_desc = field.get('description', '')
|
||
field_required = field.get('required', False)
|
||
|
||
if not field_name:
|
||
continue
|
||
|
||
prop = {
|
||
'type': type_mapping.get(field_type, 'string'),
|
||
}
|
||
|
||
if field_desc:
|
||
prop['description'] = field_desc
|
||
|
||
# 处理 object 类型的嵌套属性
|
||
if field_type == 'object' and field.get('properties'):
|
||
nested_props, nested_required = self._build_schema_properties(field['properties'])
|
||
prop['properties'] = nested_props
|
||
if nested_required:
|
||
prop['required'] = nested_required
|
||
|
||
# 处理 array 类型的元素定义
|
||
# 支持两种格式：items（标准格式）和 children（表单转 LLM 节点使用的格式）
|
||
array_items = field.get('items') or field.get('children')
|
||
if field_type == 'array' and array_items:
|
||
# 如果 children 是列表（表单字段列表），转换为 object 的 properties
|
||
if isinstance(array_items, list):
|
||
# 表单转 LLM 节点格式：children 是字段列表
|
||
nested_props, nested_required = self._build_schema_properties(array_items)
|
||
prop['items'] = {
|
||
'type': 'object',
|
||
'properties': nested_props,
|
||
}
|
||
if nested_required:
|
||
prop['items']['required'] = nested_required
|
||
elif isinstance(array_items, dict):
|
||
# 标准格式：items 是单个对象定义
|
||
items_type = array_items.get('type', 'string')
|
||
|
||
if items_type == 'object' and array_items.get('properties'):
|
||
# 数组元素是对象类型
|
||
nested_props, nested_required = self._build_schema_properties(array_items['properties'])
|
||
prop['items'] = {
|
||
'type': 'object',
|
||
'properties': nested_props,
|
||
}
|
||
if nested_required:
|
||
prop['items']['required'] = nested_required
|
||
else:
|
||
# 数组元素是基本类型
|
||
prop['items'] = {
|
||
'type': type_mapping.get(items_type, 'string'),
|
||
}
|
||
|
||
properties[field_name] = prop
|
||
|
||
if field_required:
|
||
required_fields.append(field_name)
|
||
|
||
# 调试日志
|
||
import json
|
||
logger.info(f'_build_schema_properties 结果: properties={json.dumps(properties, ensure_ascii=False)}, required={required_fields}')
|
||
|
||
return properties, required_fields
|
||
|
||
def _parse_json_from_text(
|
||
self,
|
||
content: str,
|
||
output_var: str,
|
||
total_tokens: int,
|
||
elapsed_time: int,
|
||
) -> NodeResult:
|
||
"""从文本中解析 JSON（回退方案）"""
|
||
try:
|
||
# 尝试提取 JSON
|
||
text = content.strip()
|
||
|
||
# 处理 markdown 代码块
|
||
if '```' in text:
|
||
import re
|
||
json_match = re.search(r'```(?:json)?\s*([\s\S]*?)```', text)
|
||
if json_match:
|
||
text = json_match.group(1).strip()
|
||
|
||
# 尝试找到 JSON 对象
|
||
start_idx = text.find('{')
|
||
end_idx = text.rfind('}')
|
||
if start_idx != -1 and end_idx != -1:
|
||
text = text[start_idx:end_idx + 1]
|
||
|
||
extracted_data = json.loads(text)
|
||
json_output = json.dumps(extracted_data, ensure_ascii=False, indent=2)
|
||
|
||
output_variables = {
|
||
output_var: json_output,
|
||
f'{output_var}_data': extracted_data,
|
||
f'{output_var}_tokens': total_tokens,
|
||
}
|
||
|
||
for key, value in extracted_data.items():
|
||
output_variables[f'{output_var}_{key}'] = value
|
||
|
||
logger.info(f'LLM 结构化输出成功 (文本解析), tokens: {total_tokens}')
|
||
return NodeResult(
|
||
success=True,
|
||
output=json_output,
|
||
output_variables=output_variables,
|
||
tokens_used=total_tokens,
|
||
elapsed_time=elapsed_time,
|
||
metadata={
|
||
'output_mode': 'structured',
|
||
'use_function_calling': False,
|
||
'call_method': 'Text Parsing (JSON extraction from response)',
|
||
},
|
||
)
|
||
|
||
except (json.JSONDecodeError, ValueError) as e:
|
||
logger.warning(f'JSON 解析失败: {e}, content: {content[:200]}')
|
||
return NodeResult(
|
||
success=False,
|
||
error=f'无法从 LLM 响应中解析 JSON: {str(e)}',
|
||
output=content,
|
||
output_variables={
|
||
output_var: content,
|
||
f'{output_var}_tokens': total_tokens,
|
||
},
|
||
tokens_used=total_tokens,
|
||
elapsed_time=elapsed_time,
|
||
)
|
||
|
||
async def _execute_structured_output_async(
|
||
self,
|
||
llm_service,
|
||
model_id: str,
|
||
messages: List[Dict],
|
||
temperature: float,
|
||
max_tokens: int,
|
||
output_schema: List[Dict],
|
||
output_var: str,
|
||
start_time: float,
|
||
) -> NodeResult:
|
||
"""
|
||
异步使用 Function Calling 执行结构化输出（支持多层嵌套）
|
||
"""
|
||
# 使用递归方法构建多层嵌套的 JSON Schema
|
||
properties, required_fields = self._build_schema_properties(output_schema)
|
||
|
||
if not properties:
|
||
return NodeResult(
|
||
success=False,
|
||
error='结构化输出模式需要至少定义一个输出字段',
|
||
elapsed_time=int((time.time() - start_time) * 1000),
|
||
)
|
||
|
||
# 构建 Function Calling 工具定义
|
||
tools = [{
|
||
'name': 'extract_structured_data',
|
||
'description': '从文本中提取结构化数据并返回 JSON 格式',
|
||
'parameters': {
|
||
'type': 'object',
|
||
'properties': properties,
|
||
'required': required_fields,
|
||
},
|
||
}]
|
||
|
||
# 使用 Function Calling（异步）
|
||
response = await llm_service.chat_async(
|
||
model_id=model_id,
|
||
messages=messages,
|
||
temperature=temperature,
|
||
max_tokens=max_tokens,
|
||
tools=tools,
|
||
tool_choice='required',
|
||
)
|
||
|
||
elapsed_time = int((time.time() - start_time) * 1000)
|
||
|
||
if response.has_tool_calls and response.tool_calls:
|
||
# 从 tool_calls 中提取结构化数据
|
||
extracted_data = response.tool_calls[0].arguments
|
||
|
||
# 处理双重序列化问题：如果 extracted_data 是 {"raw": "..."} 格式，提取 raw 字段
|
||
if isinstance(extracted_data, dict) and len(extracted_data) == 1 and 'raw' in extracted_data:
|
||
raw_value = extracted_data['raw']
|
||
logger.info(f'检测到 raw 包装格式，提取 raw 字段: {raw_value[:200] if isinstance(raw_value, str) else raw_value}')
|
||
if isinstance(raw_value, str):
|
||
try:
|
||
extracted_data = json.loads(raw_value)
|
||
logger.info(f'成功解析 raw 字段为 JSON 对象')
|
||
except json.JSONDecodeError as e:
|
||
logger.warning(f'raw 字段不是有效的 JSON: {e}')
|
||
# 保持原样
|
||
else:
|
||
extracted_data = raw_value
|
||
|
||
json_output = json.dumps(extracted_data, ensure_ascii=False, indent=2)
|
||
logger.info(f'LLM 结构化输出成功 (Function Calling), tokens: {response.total_tokens}')
|
||
|
||
output_variables = {
|
||
output_var: json_output,
|
||
f'{output_var}_data': extracted_data,
|
||
f'{output_var}_tokens': response.total_tokens,
|
||
f'{output_var}_raw': response.tool_calls[0].arguments, # 保留原始响应
|
||
}
|
||
|
||
for key, value in extracted_data.items():
|
||
output_variables[f'{output_var}_{key}'] = value
|
||
|
||
return NodeResult(
|
||
success=True,
|
||
output=json_output,
|
||
output_variables=output_variables,
|
||
tokens_used=response.total_tokens,
|
||
elapsed_time=elapsed_time,
|
||
metadata={
|
||
'model': response.model,
|
||
'output_mode': 'structured',
|
||
'use_function_calling': True,
|
||
'call_method': 'Function Calling (tool_choice=required)',
|
||
},
|
||
)
|
||
else:
|
||
# Function Calling 未返回结果，回退到文本解析
|
||
logger.warning('Function Calling 未返回 tool_calls，尝试解析文本响应')
|
||
return self._parse_json_from_text(
|
||
response.content, output_var, response.total_tokens, elapsed_time
|
||
)
|
||
|
||
@classmethod
|
||
def get_config_schema(cls) -> Dict[str, Any]:
|
||
"""获取配置 Schema"""
|
||
return {
|
||
'type': 'object',
|
||
'properties': {
|
||
'model_id': {
|
||
'type': 'string',
|
||
'title': '模型',
|
||
'description': '选择要使用的 LLM 模型',
|
||
},
|
||
'system_prompt': {
|
||
'type': 'string',
|
||
'title': '系统提示词',
|
||
'description': '设置 AI 的角色和行为',
|
||
'format': 'textarea',
|
||
},
|
||
'user_prompt': {
|
||
'type': 'string',
|
||
'title': '用户提示词',
|
||
'description': '支持变量引用，如 {{user_input}}',
|
||
'format': 'textarea',
|
||
},
|
||
'temperature': {
|
||
'type': 'number',
|
||
'title': '温度',
|
||
'description': '控制输出的随机性，0-2',
|
||
'default': 0.7,
|
||
'minimum': 0,
|
||
'maximum': 2,
|
||
},
|
||
'max_tokens': {
|
||
'type': 'integer',
|
||
'title': '最大 Token',
|
||
'description': '限制输出长度',
|
||
'default': 2048,
|
||
},
|
||
'use_conversation_history': {
|
||
'type': 'boolean',
|
||
'title': '使用对话历史',
|
||
'default': False,
|
||
},
|
||
'use_streaming': {
|
||
'type': 'boolean',
|
||
'title': '流式输出',
|
||
'description': '启用后将实时输出 LLM 响应',
|
||
'default': False,
|
||
},
|
||
'output_mode': {
|
||
'type': 'string',
|
||
'title': '输出模式',
|
||
'description': 'text: 普通文本输出; structured: 结构化 JSON 输出（使用 Function Calling）',
|
||
'enum': ['text', 'structured'],
|
||
'enumNames': ['文本输出', '结构化输出'],
|
||
'default': 'text',
|
||
},
|
||
'output_schema': {
|
||
'type': 'array',
|
||
'title': '输出字段定义',
|
||
'description': '结构化输出模式下，定义要提取的字段',
|
||
'items': {
|
||
'type': 'object',
|
||
'properties': {
|
||
'name': {
|
||
'type': 'string',
|
||
'title': '字段名',
|
||
'description': '英文字段名，如 country, project_name',
|
||
},
|
||
'type': {
|
||
'type': 'string',
|
||
'title': '类型',
|
||
'enum': ['string', 'number', 'integer', 'boolean', 'array', 'object'],
|
||
'default': 'string',
|
||
},
|
||
'description': {
|
||
'type': 'string',
|
||
'title': '字段描述',
|
||
'description': '描述该字段的含义，帮助 LLM 理解',
|
||
},
|
||
'required': {
|
||
'type': 'boolean',
|
||
'title': '必填',
|
||
'default': False,
|
||
},
|
||
},
|
||
'required': ['name'],
|
||
},
|
||
},
|
||
'output_variable': {
|
||
'type': 'string',
|
||
'title': '输出变量名',
|
||
'default': 'llm_response',
|
||
},
|
||
},
|
||
'required': [],
|
||
}
|