工具

test_workflow_tool.py

@@ -294,15 +294,100 @@ def get_execution_result(execution_id, headers):
         
         if output_data:
             if isinstance(output_data, dict):
-                # 尝试提取文本输出
-                text_output = (
-                    output_data.get("output") or
-                    output_data.get("text") or
-                    output_data.get("content") or
-                    output_data.get("result") or
-                    json.dumps(output_data, ensure_ascii=False, indent=2)
-                )
-                print(text_output)
+                # 如果 result 字段是字符串，尝试解析它（类似JSON节点的parse操作）
+                if "result" in output_data and isinstance(output_data["result"], str):
+                    try:
+                        # 尝试使用 ast.literal_eval 解析Python字典字符串
+                        import ast
+                        parsed_result = ast.literal_eval(output_data["result"])
+                        output_data = parsed_result
+                    except:
+                        # 如果解析失败，尝试作为JSON解析
+                        try:
+                            parsed_result = json.loads(output_data["result"])
+                            output_data = parsed_result
+                        except:
+                            pass
+                
+                # 使用类似JSON节点的extract操作来提取文本
+                def json_extract(data, path):
+                    """类似JSON节点的extract操作，使用路径提取数据"""
+                    if not path or not isinstance(data, dict):
+                        return None
+                    # 支持 $.right.right.right 格式的路径
+                    path = path.replace('$.', '').replace('$', '')
+                    keys = path.split('.')
+                    result = data
+                    for key in keys:
+                        if isinstance(result, dict) and key in result:
+                            result = result[key]
+                        else:
+                            return None
+                    return result
+                
+                # 尝试使用路径提取：递归查找 right 字段直到找到字符串
+                def extract_text_by_path(data, depth=0, max_depth=10):
+                    """递归提取嵌套在right字段中的文本"""
+                    if depth > max_depth:
+                        return None
+                    if isinstance(data, str):
+                        # 如果是字符串且不是JSON格式，返回它
+                        if len(data) > 10 and not data.strip().startswith('{') and not data.strip().startswith('['):
+                            return data
+                        return None
+                    if isinstance(data, dict):
+                        # 优先查找 right 字段
+                        if "right" in data:
+                            right_value = data["right"]
+                            # 如果 right 的值是字符串，直接返回
+                            if isinstance(right_value, str) and len(right_value) > 10:
+                                return right_value
+                            # 否则递归查找
+                            result = extract_text_by_path(right_value, depth + 1, max_depth)
+                            if result:
+                                return result
+                        # 查找其他常见的输出字段
+                        for key in ["output", "text", "content"]:
+                            if key in data:
+                                result = extract_text_by_path(data[key], depth + 1, max_depth)
+                                if result:
+                                    return result
+                        return None
+                    return None
+                
+                # 优先检查 result 字段（JSON节点提取后的结果）
+                if "result" in output_data and isinstance(output_data["result"], str):
+                    text_output = output_data["result"]
+                else:
+                    # 先尝试使用路径提取（类似JSON节点的extract操作）
+                    # 尝试多个可能的路径
+                    paths_to_try = [
+                        "right.right.right",  # 最常见的嵌套路径
+                        "right.right",
+                        "right",
+                        "output",
+                        "text",
+                        "content"
+                    ]
+                    
+                    text_output = None
+                    for path in paths_to_try:
+                        extracted = json_extract(output_data, f"$.{path}")
+                        if extracted and isinstance(extracted, str) and len(extracted) > 10:
+                            text_output = extracted
+                            break
+                    
+                    # 如果路径提取失败，使用递归提取
+                    if not text_output:
+                        text_output = extract_text_by_path(output_data)
+                
+                if text_output and isinstance(text_output, str):
+                    print(text_output)
+                    print()
+                    print_info(f"回答长度: {len(text_output)} 字符")
+                else:
+                    # 如果无法提取，显示格式化的JSON
+                    print(json.dumps(output_data, ensure_ascii=False, indent=2))
             else:
                 print(output_data)
         else: