Create app.py

app.py

@@ -0,0 +1,321 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+LoongFlow HuggingFace Spaces Demo
+展示 PES (Plan-Execute-Summary) 进化式 Agent 工作流程
+"""
+
+import gradio as gr
+import json
+import time
+import random
+from typing import List, Dict, Any
+
+# ============================================================================
+# PES 工作流程模拟
+# ============================================================================
+
+def simulate_planner(task: str) -> Dict[str, Any]:
+    """模拟 Planner 阶段的思考过程"""
+    time.sleep(0.5)
+    
+    strategies = [
+        "我将把任务分解为多个子目标，逐步实现每个目标。",
+        "首先分析任务需求，然后设计实现方案，最后进行代码编写。",
+        "采用迭代式方法，从简单版本开始，逐步添加复杂功能。",
+        "我将使用分治策略，将大问题分解为小问题解决。",
+    ]
+    
+    return {
+        "role": "Planner",
+        "thought": random.choice(strategies),
+        "plan": f"""
+## 任务分析
+- 用户需求: {task}
+
+## 执行计划
+1. 理解任务要求
+2. 设计解决方案
+3. 编写代码实现
+4. 测试验证功能
+        """.strip(),
+        "timestamp": time.strftime("%H:%M:%S")
+    }
+
+
+def simulate_executor(task: str, plan: str) -> Dict[str, Any]:
+    """模拟 Executor 阶段的执行过程"""
+    time.sleep(0.8)
+    
+    code_samples = {
+        "todo": '''```python
+# Todo List App
+class TodoList:
+    def __init__(self):
+        self.tasks = []
+    
+    def add_task(self, title):
+        self.tasks.append({"title": title, "done": False})
+    
+    def complete_task(self, index):
+        self.tasks[index]["done"] = True
+```''',
+        "file": '''```python
+# File Processor
+import os
+import shutil
+
+def process_files(input_dir, output_dir):
+    for filename in os.listdir(input_dir):
+        src = os.path.join(input_dir, filename)
+        dst = os.path.join(output_dir, filename)
+        shutil.copy2(src, dst)
+```''',
+        "default": '''```python
+# Solution Implementation
+def solve_task(task):
+    # 分析任务
+    analysis = analyze(task)
+    
+    # 设计方案
+    solution = design(analysis)
+    
+    # 实现代码
+    code = implement(solution)
+    
+    # 返回结果
+    return execute(code)
+```'''
+    }
+    
+    code = code_samples.get("default")
+    for key, c in code_samples.items():
+        if key in task.lower():
+            code = c
+            break
+    
+    return {
+        "role": "Executor",
+        "action": "编写并执行代码",
+        "code": code,
+        "result": "代码执行成功，所有测试通过 ✓",
+        "timestamp": time.strftime("%H:%M:%S")
+    }
+
+
+def simulate_summary(iteration: int, score: float) -> Dict[str, Any]:
+    """模拟 Summary 阶段的反思过程"""
+    time.sleep(0.3)
+    
+    reflections = [
+        "本次迭代成功实现了核心功能，但还有优化空间。",
+        "代码结构良好，可以进一步简化逻辑。",
+        "测试覆盖完整，建议增加边界情况处理。",
+        "性能表现良好，可以考虑添加缓存机制。",
+    ]
+    
+    improvements = [
+        "建议优化代码结构，提高可读性",
+        "可以添加更多的错误处理",
+        "考虑性能优化和资源管理",
+        "建议增加单元测试覆盖",
+    ]
+    
+    new_score = min(1.0, score + random.uniform(0.05, 0.15))
+    
+    return {
+        "role": "Summary",
+        "reflection": random.choice(reflections),
+        "improvement": random.choice(improvements),
+        "score": new_score,
+        "timestamp": time.strftime("%H:%M:%S")
+    }
+
+
+def run_pes_iteration(task: str, iteration: int, current_score: float) -> List[Dict[str, Any]]:
+    """运行一次完整的 PES 迭代"""
+    results = []
+    
+    # Phase 1: Plan
+    planner_result = simulate_planner(task)
+    results.append({
+        "phase": "📋 Plan (计划)",
+        "content": planner_result["thought"],
+        "detail": planner_result["plan"],
+        "timestamp": planner_result["timestamp"]
+    })
+    
+    # Phase 2: Execute
+    executor_result = simulate_executor(task, planner_result["plan"])
+    results.append({
+        "phase": "⚡ Execute (执行)",
+        "content": executor_result["action"],
+        "detail": f"{executor_result['code']}\n\n**执行结果**: {executor_result['result']}",
+        "timestamp": executor_result["timestamp"]
+    })
+    
+    # Phase 3: Summary
+    summary_result = simulate_summary(iteration, current_score)
+    results.append({
+        "phase": "📊 Summary (总结)",
+        "content": summary_result["reflection"],
+        "detail": f"**改进建议**: {summary_result['improvement']}\n\n**当前分数**: {summary_result['score']:.2f}",
+        "timestamp": summary_result["timestamp"]
+    })
+    
+    return results, summary_result["score"]
+
+
+# ============================================================================
+# Gradio UI
+# ============================================================================
+
+def create_demo():
+    """创建 Gradio 界面"""
+    
+    with gr.Blocks(title="LoongFlow PES Demo", theme=gr.themes.Soft()) as demo:
+        gr.Markdown("""
+        # 🐉 LoongFlow PES Agent Demo
+        
+        **LoongFlow** 是一个进化式 Agent 开发框架，采用 **PES (Plan-Execute-Summary)** 思考范式。
+        
+        这个 Demo 展示了一个 Agent 如何通过计划-执行-总结的循环来解决问题，并逐步改进解决方案。
+        
+        ---
+        
+        ### PES 工作流程
+        
+        ```
+        ┌─────────────┐     ┌─────────────┐     ┌─────────────┐
+        │    Plan     │────▶│   Execute   │────▶│   Summary   │
+        │   (计划)    │     │   (执行)    │     │   (总结)    │
+        └─────────────┘     └─────────────┘     └─────────────┘
+              │                                           │
+              │              迭代改进 ◀───────────────────┘
+              │                    
+        ┌─────────────┐           
+        │   目标达成   │           
+        └─────────────┘           
+        ```
+        """)
+        
+        with gr.Row():
+            with gr.Column(scale=2):
+                task_input = gr.Textbox(
+                    label="输入任务描述",
+                    placeholder="例如: 帮我写一个待办事项应用 / 创建一个文件处理工具",
+                    lines=3
+                )
+                
+                with gr.Row():
+                    max_iterations = gr.Slider(
+                        minimum=1, maximum=10, value=5, step=1,
+                        label="最大迭代次数"
+                    )
+                    target_score = gr.Slider(
+                        minimum=0.5, maximum=1.0, value=0.85, step=0.05,
+                        label="目标分数"
+                    )
+                
+                run_btn = gr.Button("🚀 开始执行任务", variant="primary")
+            
+            with gr.Column(scale=1):
+                status_output = gr.JSON(
+                    label="执行状态",
+                    visible=False
+                )
+        
+        with gr.Row():
+            score_chart = gr.LinePlot(
+                label="分数演进",
+                x="iteration",
+                y="score",
+                height=300
+            )
+        
+        with gr.Accordion("📝 迭代详情", open=True):
+            iteration_output = gr.JSON(
+                label="迭代历史",
+                elem_id="iteration-output"
+            )
+        
+        # 状态变量
+        iterations_data = []
+        
+        def run_task(task: str, max_iter: int, target: float):
+            if not task.strip():
+                yield {"error": "请输入任务描述"}, [], gr.update()
+                return
+            
+            iterations_data.clear()
+            yield {"status": "开始执行任务...", "task": task}, [], gr.update()
+            
+            current_score = 0.0
+            all_results = []
+            
+            for i in range(1, max_iter + 1):
+                yield {
+                    "status": f"正在执行第 {i} 次迭代...",
+                    "iteration": i,
+                    "score": current_score
+                }, [], gr.update(visible=True)
+                
+                results, current_score = run_pes_iteration(task, i, current_score)
+                all_results.extend(results)
+                
+                # 更新图表数据
+                iterations_data.append({"iteration": i, "score": current_score})
+                
+                yield {
+                    "status": f"第 {i} 次迭代完成",
+                    "iteration": i,
+                    "score": current_score
+                }, all_results, gr.update(
+                    value=iterations_data,
+                    visible=True
+                )
+                
+                # 检查是否达到目标
+                if current_score >= target:
+                    break
+                
+                time.sleep(0.5)
+            
+            final_status = {
+                "status": "任务完成" if current_score >= target else "达到最大迭代",
+                "final_score": current_score,
+                "iterations": len(iterations_data)
+            }
+            
+            yield final_status, all_results, gr.update(
+                value=iterations_data,
+                visible=True
+            )
+        
+        run_btn.click(
+            fn=run_task,
+            inputs=[task_input, max_iterations, target_score],
+            outputs=[status_output, iteration_output, score_chart]
+        )
+        
+        gr.Markdown("""
+        ---
+        
+        ### 关于 LoongFlow
+        
+        LoongFlow 是一个面向复杂任务的进化式 Agent 框架，特别适用于：
+        
+        - 📐 **数学推理**: 开放式数学问题求解
+        - 🤖 **机器学习**: AutoML 和算法优化
+        - 💻 **代码生成**: 复杂编程任务
+        - 🔬 **科学研究**: 实验设计和分析
+        
+        了解更多: [GitHub](https://github.com/your-repo/loongflow)
+        """)
+    
+    return demo
+
+
+if __name__ == "__main__":
+    demo = create_demo()
+    demo.launch(server_name="0.0.0.0", server_port=7860)