Files
aiagent/test_output_variable_extraction.py

67 lines
2.1 KiB
Python
Raw Permalink Normal View History

2026-01-22 09:59:02 +08:00
#!/usr/bin/env python3
"""
测试output变量提取逻辑
"""
import sys
sys.path.insert(0, 'backend')
from app.services.workflow_engine import WorkflowEngine
# 模拟llm-format节点的输入数据
input_data = {
'right': {
'right': {
'right': '是的,我记得!根据我们之前的对话,你告诉我你的名字叫"老七"。我会在本次对话中记住这个名字,以便更好地为你提供帮助。如果你希望我用其他称呼,也可以随时告诉我。',
'query': '你还记得我的名字吗?'
},
'memory': {
'conversation_history': [],
'user_profile': {},
'context': {}
},
'query': '你还记得我的名字吗?'
},
'memory': {
'conversation_history': [],
'user_profile': {},
'context': {}
},
'query': '你还记得我的名字吗?'
}
# 创建WorkflowEngine实例
engine = WorkflowEngine("test", {"nodes": [], "edges": []})
# 测试_get_nested_value方法
print("测试_get_nested_value方法:")
value1 = engine._get_nested_value(input_data, 'output')
print(f" _get_nested_value(input_data, 'output'): {value1}")
# 测试output变量提取逻辑
print("\n测试output变量提取逻辑:")
right_value = input_data.get('right')
print(f" right_value类型: {type(right_value)}")
print(f" right_value: {str(right_value)[:100]}")
if right_value is not None:
if isinstance(right_value, str):
value = right_value
print(f" ✅ 从right字段字符串提取: {value[:100]}")
elif isinstance(right_value, dict):
current = right_value
depth = 0
while isinstance(current, dict) and depth < 10:
if 'right' in current:
current = current['right']
depth += 1
if isinstance(current, str):
value = current
print(f" ✅ 从right字段嵌套{depth}层)提取: {value[:100]}")
break
else:
break
else:
print(f" ❌ 无法提取字符串值")
print("\n测试完成")