Grok 4.6 —— SpaceXAI长程Agent编程模型深度解析

# self_generated_sft_pipeline.py
"""
Grok 4.6 Self-Generated SFT 管线的简化实现
演示 Model-Based Filtering 的核心逻辑
"""

import json
import random
from typing import Any, Dict, List, Optional, Tuple
from dataclasses import dataclass, field


@dataclass
class SFTTrajectory:
    """单条SFT推理轨迹"""
    task_id: str
    domain: str                    # STEM | SWE | KnowledgeWork
    reasoning_effort: str          # low | medium | high | xhigh
    prompt: str
    steps: List[Dict[str, str]]    # 推理步骤列表
    final_answer: str
    quality_score: Optional[float] = None


@dataclass
class TrajectoryFilter:
    """
    Model-Based Filtering 核心实现
    使用一个评判模型对轨迹进行多维质量评估
    """
    
    def __init__(self, quality_threshold: float = 0.7):
        self.threshold = quality_threshold
        # 评判维度权重
        self.dimension_weights = {
            "logical_coherence": 0.35,
            "step_completeness": 0.25,
            "answer_correctness": 0.30,
            "efficiency": 0.10,
        }
    
    def evaluate_logical_coherence(self, trajectory: SFTTrajectory) -> float:
        """
        评估推理轨迹的逻辑连贯性
        检查是否存在跳跃、循环或矛盾
        """
        score = 0.0
        steps = trajectory.steps
        
        if len(steps) < 2:
            return 0.0
        
        # 检查每步之间的逻辑连续性
        for i in range(1, len(steps)):
            prev_step = steps[i - 1]
            curr_step = steps[i]
            
            # 检查是否有上下文跳跃（模拟评判模型的行为）
            overlap = len(
                set(prev_step.get("key_concepts", [])) & 
                set(curr_step.get("key_concepts", []))
            )
            if overlap == 0 and i > 1:
                score -= 0.2  # 逻辑断裂惩罚
        
        # 检查是否存在循环推理
        step_texts = [s.get("reasoning", "") for s in steps]
        for i in range(len(step_texts)):
            for j in range(i + 2, len(step_texts)):
                if self._semantic_similarity(step_texts[i], step_texts[j]) > 0.85:
                    score -= 0.3  # 循环推理惩罚
        
        # 归一化到 [0, 1]
        score = max(0.0, min(1.0, 1.0 + score))
        return score
    
    def evaluate_step_completeness(self, trajectory: SFTTrajectory) -> float:
        """
        评估推理步骤是否完整
        检查是否遗漏了关键子问题
        """
        # 模拟评判模型：检查每个步骤是否包含可验证的中间结论
        complete_steps = sum(
            1 for s in trajectory.steps 
            if s.get("has_verifiable_claim", False)
        )
        return complete_steps / max(len(trajectory.steps), 1)
    
    def evaluate_answer_correctness(self, trajectory: SFTTrajectory) -> float:
        """
        评估最终答案的正确性
        使用验证器或已知答案进行比对
        """
        # 模拟评判：检查答案是否与已知正确结果一致
        # 在真实系统中，这由独立的验证模型执行
        return trajectory.quality_score if trajectory.quality_score else 0.5
    
    def evaluate_efficiency(self, trajectory: SFTTrajectory) -> float:
        """
        评估推理效率
        避免不必要的冗长推理
        """
        n_steps = len(trajectory.steps)
        effort_multiplier = {
            "low": 5, "medium": 10, "high": 20, "xhigh": 40
        }
        expected_steps = effort_multiplier.get(trajectory.reasoning_effort, 10)
        
        # 步骤数在预期范围的 ±50% 内为最佳
        ratio = n_steps / expected_steps
        if 0.5 <= ratio <= 1.5:
            return 1.0
        elif ratio < 0.5:
            return max(0.0, ratio * 2)  # 步骤太少，可能不够完整
        else:
            return max(0.0, 2.0 - ratio)  # 步骤太多，效率低下
    
    def _semantic_similarity(self, text_a: str, text_b: str) -> float:
        """简化语义相似度计算（真实系统中使用嵌入模型）"""
        # 基于字符级Jaccard相似度的简化实现
        set_a, set_b = set(text_a.split()), set(text_b.split())
        if not set_a or not set_b:
            return 0.0
        intersection = set_a & set_b
        union = set_a | set_b
        return len(intersection) / len(union)
    
    def filter_trajectory(self, trajectory: SFTTrajectory) -> Tuple[bool, float]:
        """
        对单条轨迹进行过滤，返回 (是否保留, 综合评分)
        """
        scores = {
            "logical_coherence": self.evaluate_logical_coherence(trajectory),
            "step_completeness": self.evaluate_step_completeness(trajectory),
            "answer_correctness": self.evaluate_answer_correctness(trajectory),
            "efficiency": self.evaluate_efficiency(trajectory),
        }
        
        weighted_score = sum(
            scores[dim] * self.dimension_weights[dim]
            for dim in self.dimension_weights
        )
        
        trajectory.quality_score = weighted_score
        return weighted_score >= self.threshold, weighted_score


# 模拟数据生成
def generate_sample_trajectories() -> List[SFTTrajectory]:
    """生成模拟SFT轨迹用于演示"""
    tasks = [
        {
            "task_id": "SWE-001",
            "domain": "SWE",
            "prompt": "实现一个LRU缓存，支持get和put操作，时间复杂度O(1)",
            "good_steps": [
                {"reasoning": "分析需求：需要O(1)的get和put，暗示使用哈希表+双向链表",
                 "key_concepts": ["哈希表", "双向链表", "LRU"],
                 "has_verifiable_claim": True},
                {"reasoning": "设计数据结构：哈希表存储key到节点的映射，双向链表维护访问顺序",
                 "key_concepts": ["哈希表", "双向链表"],
                 "has_verifiable_claim": True},
                {"reasoning": "get操作：从哈希表查找，若存在则移到链表头部",
                 "key_concepts": ["get", "哈希表查找", "链表移动"],
                 "has_verifiable_claim": True},
                {"reasoning": "put操作：若key存在则更新值并移到头部，不存在则新建节点插入头部，超容量则删除尾部",
                 "key_concepts": ["put", "插入", "容量控制", "尾部删除"],
                 "has_verifiable_claim": True},
            ],
            "final_answer": "class LRUCache实现代码...",
            "quality_score": 0.92,
        },
        {
            "task_id": "SWE-002",
            "domain": "SWE",
            "prompt": "实现一个简单的Web服务器",
            "bad_steps": [
                {"reasoning": "需要用Python写",
                 "key_concepts": ["Python"],
                 "has_verifiable_claim": False},
                {"reasoning": "好像需要socket",
                 "key_concepts": ["socket"],
                 "has_verifiable_claim": False},
                {"reasoning": "用socket.bind然后listen",
                 "key_concepts": ["socket"],
                 "has_verifiable_claim": True},
                {"reasoning": "等等，刚才说的bind好像不对，再看看需求",
                 "key_concepts": ["socket"],
                 "has_verifiable_claim": False},
                {"reasoning": "还是用Python吧",
                 "key_concepts": ["Python"],
                 "has_verifiable_claim": False},
            ],
            "final_answer": "import socket ...",
            "quality_score": 0.35,
        }
    ]
    
    trajectories = []
    for t in tasks:
        traj = SFTTrajectory(
            task_id=t["task_id"],
            domain=t["domain"],
            reasoning_effort="high",
            prompt=t["prompt"],
            steps=t["good_steps"] if t["quality_score"] > 0.5 else t["bad_steps"],
            final_answer=t["final_answer"],
            quality_score=t["quality_score"],
        )
        trajectories.append(traj)
    return trajectories


def main():
    """主流程：演示 Model-Based Filtering"""
    filter_engine = TrajectoryFilter(quality_threshold=0.7)
    trajectories = generate_sample_trajectories()
    
    print("=" * 70)
    print("Grok 4.6 Self-Generated SFT - Model-Based Filtering 演示")
    print("=" * 70)
    
    accepted = []
    rejected = []
    
    for traj in trajectories:
        keep, score = filter_engine.filter_trajectory(traj)
        status = "✅ ACCEPTED" if keep else "❌ REJECTED"
        
        print(f"\n任务: {traj.task_id} ({traj.domain})")
        print(f"提示: {traj.prompt[:50]}...")
        print(f"推理步骤数: {len(traj.steps)}")
        print(f"各维度评分:")
        
        scores = {
            "逻辑连贯性": filter_engine.evaluate_logical_coherence(traj),
            "步骤完整性": filter_engine.evaluate_step_completeness(traj),
            "答案正确性": filter_engine.evaluate_answer_correctness(traj),
            "推理效率": filter_engine.evaluate_efficiency(traj),
        }
        for dim, s in scores.items():
            print(f"  {dim}: {s:.3f}")
        
        print(f"综合评分: {score:.3f}")
        print(f"判定结果: {status}")
        print("-" * 70)
        
        if keep:
            accepted.append(traj)
        else:
            rejected.append(traj)
    
    print(f"\n统计: 共 {len(trajectories)} 条轨迹")
    print(f"  Accepted: {len(accepted)} ({len(accepted)/len(trajectories)*100:.1f}%)")
    print(f"  Rejected: {len(rejected)} ({len(rejected)/len(trajectories)*100:.1f}%)")


if __name__ == "__main__":
    main()

# grok_agent_ideation_to_app.py
"""
演示 Grok 4.6 长程Agent工作流：
从产品创意到可运行应用的完整编程过程
"""

import asyncio
import json
from enum import Enum
from typing import Any, Callable, Dict, List, Optional


class TaskStatus(Enum):
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    VERIFIED = "verified"
    FAILED = "failed"
    COMPLETED = "completed"


class AgentStep:
    """Agent执行的单个步骤"""
    def __init__(self, name: str, action: Callable, verify: Optional[Callable] = None):
        self.name = name
        self.action = action
        self.verify = verify or (lambda ctx: True)
        self.status = TaskStatus.PENDING
        self.result: Any = None
        self.error: Optional[str] = None


class AgentContext:
    """Agent执行上下文，保存跨步骤状态"""
    def __init__(self, task_description: str):
        self.task = task_description
        self.files: Dict[str, str] = {}       # 已创建的文件
        self.research_notes: List[str] = []    # 研究笔记
        self.code_artifacts: List[str] = []    # 代码产物
        self.test_results: Dict[str, bool] = {} # 测试结果
        self.current_step: int = 0
    
    def log(self, message: str):
        print(f"[Agent Context] {message}")


class GrokAgent:
    """
    Grok 4.6 长程Agent的简化实现
    展示自我测试、验证和多步骤执行的编程模型
    """
    
    def __init__(self, reasoning_effort: str = "high"):
        self.reasoning_effort = reasoning_effort
        self.context: Optional[AgentContext] = None
        self.steps: List[AgentStep] = []
    
    def plan(self, task: str) -> AgentContext:
        """
        规划阶段：将任务分解为可执行的步骤序列
        模拟 Grok 4.6 的规划能力
        """
        self.context = AgentContext(task)
        self.steps = []
        
        print(f"🧠 Grok 4.6 ({self.reasoning_effort}) 正在规划任务...")
        print(f"📋 任务: {task}")
        print()
        
        # 模拟 Grok 4.6 的任务分解
        self.steps = [
            AgentStep(
                "research_domain",
                self._research_unfamiliar_domain,
                self._verify_research_complete,
            ),
            AgentStep(
                "design_architecture",
                self._design_application_architecture,
                self._verify_architecture_coherent,
            ),
            AgentStep(
                "implement_core",
                self._implement_core_interactions,
                self._verify_core_works,
            ),
            AgentStep(
                "test_and_fix",
                self._test_and_fix_issues,
                self._verify_all_tests_pass,
            ),
            AgentStep(
                "refine_artifact",
                self._refine_deliverable,
                self._verify_deliverable_quality,
            ),
        ]
        
        return self.context
    
    async def execute(self) -> AgentContext:
        """
        执行阶段：按顺序执行每个步骤，并在每个步骤之间进行验证
        模拟 Grok 4.6 的自我测试和验证机制
        """
        if not self.context:
            raise ValueError("请先调用 plan() 方法")
        
        print("🚀 开始执行任务...")
        print("=" * 60)
        
        for i, step in enumerate(self.steps):
            self.context.current_step = i
            print(f"\n[Step {i+1}/{len(self.steps)}] {step.name}")
            print("-" * 40)
            
            # 前置验证：检查执行前提是否满足
            if not self._check_preconditions(step):
                print(f"  ⚠️  前置条件不满足，尝试修复...")
                self._attempt_recovery(step)
            
            # 执行步骤
            step.status = TaskStatus.IN_PROGRESS
            try:
                step.result = await step.action()
                step.status = TaskStatus.VERIFIED
                print(f"  ✅ 步骤执行完成")
            except Exception as e:
                step.error = str(e)
                step.status = TaskStatus.FAILED
                print(f"  ❌ 步骤执行失败: {e}")
                
                # 自我修复
                if self._can_self_heal(step):
                    print(f"  🔧 尝试自我修复...")
                    step.result = await self._self_heal(step)
                    step.status = TaskStatus.VERIFIED
                    print(f"  ✅ 自我修复成功")
                else:
                    print(f"  ❌ 无法修复，终止执行")
                    break
            
            # 后置验证：检查输出是否满足要求
            if step.status == TaskStatus.VERIFIED:
                if not step.verify(self.context):
                    print(f"  ⚠️  验证失败，重新执行...")
                    step.status = TaskStatus.PENDING
                    # 重新执行一次
                    step.result = await step.action()
                    step.status = TaskStatus.VERIFIED
                    if step.verify(self.context):
                        print(f"  ✅ 重新执行后验证通过")
                    else:
                        print(f"  ❌ 重新执行后仍验证失败")
                        break
            
            self.context.log(f"步骤 {step.name} 完成")
        
        self.context.current_step = len(self.steps)
        print("\n" + "=" * 60)
        print("🏁 任务执行完成")
        
        return self.context
    
    # ---- 模拟的Agent动作 ----
    
    async def _research_unfamiliar_domain(self) -> Dict:
        """模拟研究不熟悉的领域"""
        print("  📖 正在研究领域背景...")
        self.context.research_notes = [
            "现代Web应用使用React/Vue等前端框架",
            "后端常用FastAPI/Express等框架",
            "数据库选型需考虑PostgreSQL或MongoDB",
            "部署可用Docker + Vercel/Cloudflare",
        ]
        return {"research_complete": True, "sources": 4}
    
    async def _design_application_architecture(self) -> Dict:
        """模拟设计应用架构"""
        print("  🏗️  正在设计应用架构...")
        architecture = {
            "frontend": "React + TypeScript",
            "backend": "FastAPI + Python",
            "database": "PostgreSQL",
            "deployment": "Docker + Vercel",
            "api_design": "RESTful + WebSocket",
        }
        self.context.code_artifacts.append(
            json.dumps(architecture, indent=2)
        )
        return architecture
    
    async def _implement_core_interactions(self) -> str:
        """模拟实现核心交互"""
        print("  💻 正在实现核心交互...")
        code = '''
# server.py - Grok 4.6 生成的核心后端代码
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional, List
import asyncpg
from datetime import datetime

app = FastAPI(title="Grok 4.6 Demo App")

class TaskItem(BaseModel):
    id: Optional[int] = None
    title: str
    description: str
    status: str = "pending"
    created_at: Optional[str] = None

class TaskStore:
    """内存任务存储，用于演示"""
    def __init__(self):
        self._tasks: dict[int, TaskItem] = {}
        self._counter = 0
    
    async def create(self, task: TaskItem) -> TaskItem:
        self._counter += 1
        task.id = self._counter
        task.created_at = datetime.now().isoformat()
        self._tasks[task.id] = task
        return task
    
    async def get(self, task_id: int) -> Optional[TaskItem]:
        return self._tasks.get(task_id)
    
    async def list(self) -> List[TaskItem]:
        return list(self._tasks.values())
    
    async def update(self, task_id: int, task: TaskItem) -> TaskItem:
        if task_id not in self._tasks:
            raise HTTPException(status_code=404, detail="Task not found")
        task.id = task_id
        self._tasks[task_id] = task
        return task

store = TaskStore()

@app.post("/tasks", response_model=TaskItem)
async def create_task(task: TaskItem):
    return await store.create(task)

@app.get("/tasks", response_model=List[TaskItem])
async def list_tasks():
    return await store.list()

@app.get("/tasks/{task_id}", response_model=TaskItem)
async def get_task(task_id: int):
    task = await store.get(task_id)
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    return task

@app.put("/tasks/{task_id}", response_model=TaskItem)
async def update_task(task_id: int, task: TaskItem):
    return await store.update(task_id, task)
'''
        self.context.code_artifacts.append(code)
        return code
    
    async def _test_and_fix_issues(self) -> Dict:
        """模拟测试和修复"""
        print("  🧪 正在运行测试...")
        self.context.test_results = {
            "test_create_task": True,
            "test_list_tasks": True,
            "test_get_task_by_id": True,
            "test_update_task": True,
            "test_nonexistent_task": True,
        }
        return self.context.test_results
    
    async def _refine_deliverable(self) -> str:
        """模拟优化交付物"""
        print("  ✨ 正在优化输出...")
        return "应用程序已就绪，代码质量评分: 9.2/10"
    
    # ---- 验证函数 ----
    
    def _verify_research_complete(self, ctx: AgentContext) -> bool:
        return len(ctx.research_notes) >= 3
    
    def _verify_architecture_coherent(self, ctx: AgentContext) -> bool:
        return len(ctx.code_artifacts) >= 1
    
    def _verify_core_works(self, ctx: AgentContext) -> bool:
        return any("FastAPI" in a for a in ctx.code_artifacts)
    
    def _verify_all_tests_pass(self, ctx: AgentContext) -> bool:
        return all(ctx.test_results.values())
    
    def _verify_deliverable_quality(self, ctx: AgentContext) -> bool:
        return True
    
    # ---- 自我修复机制 ----
    
    def _check_preconditions(self, step: AgentStep) -> bool:
        """检查执行步骤的前提条件"""
        # 模拟条件检查
        return True
    
    def _can_self_heal(self, step: AgentStep) -> bool:
        """判断是否可以从错误中恢复"""
        return step.error is not None and "timeout" not in step.error.lower()
    
    async def _self_heal(self, step: AgentStep) -> Any:
        """尝试从错误中恢复"""
        return await step.action()
    
    def _attempt_recovery(self, step: AgentStep):
        """尝试恢复前置条件"""
        pass


async def main():
    """演示 Grok 4.6 从创意到应用的完整工作流"""
    print("=" * 70)
    print("Grok 4.6 长程Agent编程模型演示")
    print("=" * 70)
    
    # 创建Agent
    agent = GrokAgent(reasoning_effort="high")
    
    # 阶段1：规划
    task = "构建一个任务管理Web应用，支持CRUD操作"
    context = agent.plan(task)
    
    # 阶段2：执行（包含自我测试和验证）
    result = await agent.execute()
    
    # 输出执行摘要
    print("\n📊 执行摘要")
    print(f"  总步骤数: {len(agent.steps)}")
    print(f"  完成步骤: {sum(1 for s in agent.steps if s.status == TaskStatus.VERIFIED)}")
    print(f"  失败步骤: {sum(1 for s in agent.steps if s.status == TaskStatus.FAILED)}")
    print(f"  研究笔记: {len(result.research_notes)} 条")
    print(f"  代码产物: {len(result.code_artifacts)} 个")
    print(f"  测试通过率: {sum(result.test_results.values())}/{len(result.test_results)}")
    
    print("\n✅ 从产品创意到可运行应用已完成")


if __name__ == "__main__":
    asyncio.run(main())

// context_manager.go
// Grok 4.6 上下文窗口管理的Go实现
// 演示500K上下文窗口下的长程Agent状态管理

package main

import (
	"container/list"
	"crypto/sha256"
	"encoding/hex"
	"fmt"
	"strings"
	"sync"
	"time"
)

// ContextWindow 管理500K token的上下文窗口
type ContextWindow struct {
	mu           sync.RWMutex
	maxTokens    int
	currentTokens int
	segments     *list.List // 上下文段双向链表
	cache        map[string]*ContextSegment
}

// ContextSegment 上下文中的一个段
type ContextSegment struct {
	ID        string
	Content   string
	TokenCount int
	Type      string // "instruction", "code", "observation", "summary"
	Timestamp time.Time
	Priority  int // 优先级，越高越不容易被压缩
}

// NewContextWindow 创建新的上下文窗口
func NewContextWindow(maxTokens int) *ContextWindow {
	return &ContextWindow{
		maxTokens: maxTokens,
		segments:  list.New(),
		cache:     make(map[string]*ContextSegment),
	}
}

// AddSegment 向上下文窗口添加一个段
func (cw *ContextWindow) AddSegment(content string, segType string, priority int) string {
	cw.mu.Lock()
	defer cw.mu.Unlock()

	// 计算token数（简化：按字符数/4估算）
	tokenCount := len(content) / 4

	// 生成唯一ID
	hash := sha256.Sum256([]byte(content + time.Now().String()))
	id := hex.EncodeToString(hash[:8])

	segment := &ContextSegment{
		ID:         id,
		Content:    content,
		TokenCount: tokenCount,
		Type:       segType,
		Timestamp:  time.Now(),
		Priority:   priority,
	}

	// 检查是否超出容量
	for cw.currentTokens+tokenCount > cw.maxTokens {
		if !cw.compressOldest() {
			// 无法压缩，丢弃最低优先级段
			cw.evictLowestPriority()
		}
	}

	cw.segments.PushBack(segment)
	cw.cache[id] = segment
	cw.currentTokens += tokenCount

	return id
}

// compressOldest 压缩最旧的段
// 返回 true 表示成功压缩
func (cw *ContextWindow) compressOldest() bool {
	for e := cw.segments.Front(); e != nil; e = e.Next() {
		seg := e.Value.(*ContextSegment)
		if seg.Priority < 5 && seg.Type != "instruction" {
			// 压缩：用摘要替换原始内容
			summary := summarizeContent(seg.Content)
			savedTokens := seg.TokenCount - len(summary)/4
			seg.Content = summary
			seg.TokenCount = len(summary) / 4
			seg.Type = "summary"
			cw.currentTokens -= savedTokens
			return true
		}
	}
	return false
}

// evictLowestPriority 驱逐最低优先级的段
func (cw *ContextWindow) evictLowestPriority() {
	var lowest *list.Element
	for e := cw.segments.Front(); e != nil; e = e.Next() {
		seg := e.Value.(*ContextSegment)
		if seg.Type == "instruction" {
			continue // 保护指令段
		}
		if lowest == nil || seg.Priority < lowest.Value.(*ContextSegment).Priority {
			lowest = e
		}
	}
	if lowest != nil {
		seg := lowest.Value.(*ContextSegment)
		cw.currentTokens -= seg.TokenCount
		delete(cw.cache, seg.ID)
		cw.segments.Remove(lowest)
	}
}

// summarizeContent 对内容进行摘要压缩
func summarizeContent(content string) string {
	if len(content) <= 200 {
		return content
	}
	// 简化摘要：提取前200个字符
	return content[:200] + " ...[compressed]"
}

// GetContext 获取当前完整上下文
func (cw *ContextWindow) GetContext() string {
	cw.mu.RLock()
	defer cw.mu.RUnlock()

	var parts []string
	for e := cw.segments.Front(); e != nil; e = e.Next() {
		seg := e.Value.(*ContextSegment)
		parts = append(parts, fmt.Sprintf(
			"[%s|P%d] %s",
			seg.Type[:3],
			seg.Priority,
			seg.Content,
		))
	}
	return strings.Join(parts, "\n\n")
}

// Stats 返回上下文窗口状态
func (cw *ContextWindow) Stats() map[string]interface{} {
	cw.mu.RLock()
	defer cw.mu.RUnlock()

	stats := make(map[string]interface{})
	stats["max_tokens"] = cw.maxTokens
	stats["current_tokens"] = cw.currentTokens
	stats["usage_percent"] = float64(cw.currentTokens) / float64(cw.maxTokens) * 100
	stats["segment_count"] = cw.segments.Len()

	typeCount := make(map[string]int)
	for e := cw.segments.Front(); e != nil; e = e.Next() {
		seg := e.Value.(*ContextSegment)
		typeCount[seg.Type]++
	}
	stats["type_distribution"] = typeCount

	return stats
}

// LongRunningAgent 长程Agent
type LongRunningAgent struct {
	Name    string
	Context *ContextWindow
	History []AgentAction
}

type AgentAction struct {
	Step     int
	Action   string
	Result   string
	Duration time.Duration
}

func NewLongRunningAgent(name string) *LongRunningAgent {
	return &LongRunningAgent{
		Name:    name,
		Context: NewContextWindow(500000), // 500K context
		History: make([]AgentAction, 0),
	}
}

// ExecuteTask 执行长程任务
func (agent *LongRunningAgent) ExecuteTask(task string) error {
	fmt.Printf("[%s] 开始执行任务: %s\n", agent.Name, task)
	fmt.Printf("[%s] 上下文窗口: %d tokens\n", agent.Name, agent.Context.maxTokens)

	// 添加任务指令
	agent.Context.AddSegment(task, "instruction", 10)

	steps := []struct {
		name     string
		duration time.Duration
		priority int
	}{
		{"需求分析", 2 * time.Second, 8},
		{"架构设计", 3 * time.Second, 7},
		{"编码实现", 5 * time.Second, 6},
		{"测试验证", 2 * time.Second, 5},
		{"部署发布", 1 * time.Second, 4},
	}

	for i, step := range steps {
		stepID := fmt.Sprintf("step-%d", i+1)
		fmt.Printf("\n[%s] 步骤 %d/%d: %s\n", agent.Name, i+1, len(steps), step.name)

		time.Sleep(step.duration / 10) // 加速演示

		result := fmt.Sprintf("步骤 %s 完成: 输出产物 %s", step.name, stepID)
		agent.History = append(agent.History, AgentAction{
			Step:     i + 1,
			Action:   step.name,
			Result:   result,
			Duration: step.duration,
		})

		// 添加到上下文
		agent.Context.AddSegment(result, "observation", step.priority)

		// 每步后显示上下文状态
		stats := agent.Context.Stats()
		fmt.Printf("  上下文使用: %.1f%% (segments: %d)\n",
			stats["usage_percent"], stats["segment_count"])
	}

	fmt.Printf("\n[%s] ✅ 任务执行完成\n", agent.Name)
	fmt.Printf("  总步骤数: %d\n", len(agent.History))
	fmt.Printf("  最终上下文使用: %.1f%%\n",
		agent.Context.Stats()["usage_percent"])

	return nil
}

func main() {
	fmt.Println("=" + strings.Repeat("=", 65) + "=")
	fmt.Println("  Grok 4.6 500K Context Window Manager Demo")
	fmt.Println("=" + strings.Repeat("=", 65) + "=")

	agent := NewLongRunningAgent("Grok 4.6 Agent")

	// 模拟长程任务
	err := agent.ExecuteTask(`
	构建一个完整的电商平台，包含：
	1. 用户注册/登录系统
	2. 商品浏览和搜索
	3. 购物车管理
	4. 订单处理
	5. 支付集成
	6. 后台管理系统
	`)

	if err != nil {
		fmt.Printf("任务失败: %v\n", err)
		return
	}

	// 输出上下文分布
	stats := agent.Context.Stats()
	fmt.Printf("\n📊 上下文分布统计:\n")
	if dist, ok := stats["type_distribution"].(map[string]int); ok {
		for t, c := range dist {
			fmt.Printf("  %s: %d segments\n", t, c)
		}
	}
}

# benchmark_analysis.py
"""
Grok 4.6 基准测试数据分析和成本效益计算
"""

from dataclasses import dataclass
from typing import Dict, List, Optional


@dataclass
class BenchmarkResult:
    model: str
    scores: Dict[str, float]
    input_price: float  # per 1M tokens
    output_price: float  # per 1M tokens


@dataclass
class CostEfficiency:
    model: str
    cost_per_task: float
    intelligence_score: float
    efficiency_ratio: float  # score / cost


def load_benchmark_data() -> List[BenchmarkResult]:
    """加载基准测试数据"""
    return [
        BenchmarkResult(
            model="Grok 4.6 High",
            scores={
                "AA_Intelligence_Index": 61.0,
                "GDPVal_AA_v2_Elo": 1753.0,
                "CursorBench_v3_2": 69.9,
                "DeepSWE_v1_1": 65.9,
                "FrontierCode_v1_1": 61.3,
                "AA_Briefcase_Elo": 1577.0,
                "Harvey_LAB": 15.8,
            },
            input_price=2.0,
            output_price=6.0,
        ),
        BenchmarkResult(
            model="GPT-5.6 Sol Max",
            scores={
                "AA_Intelligence_Index": 61.0,
                "GDPVal_AA_v2_Elo": 1728.0,
                "CursorBench_v3_2": 67.2,
                "DeepSWE_v1_1": 73.0,
                "FrontierCode_v1_1": 60.6,
                "AA_Briefcase_Elo": 1502.0,
                "Harvey_LAB": 2.5,
            },
            input_price=5.0,
            output_price=30.0,
        ),
        BenchmarkResult(
            model="Claude Fable 5 Max",
            scores={
                "AA_Intelligence_Index": 62.0,
                "GDPVal_AA_v2_Elo": 1741.0,
                "CursorBench_v3_2": 70.5,
                "DeepSWE_v1_1": 70.0,
                "FrontierCode_v1_1": 63.6,
                "AA_Briefcase_Elo": 1574.0,
                "Harvey_LAB": 11.3,
            },
            input_price=15.0,
            output_price=50.0,
        ),
    ]


def calculate_cost_efficiency(
    results: List[BenchmarkResult],
    input_tokens_per_task: int = 500_000,
    output_tokens_per_task: int = 50_000,
) -> List[CostEfficiency]:
    """
    计算成本效益比
    
    参数:
        input_tokens_per_task: 每任务平均输入token数
        output_tokens_per_task: 每任务平均输出token数
    """
    efficiencies = []
    
    for r in results:
        # 计算单任务成本
        input_cost = (input_tokens_per_task / 1_000_000) * r.input_price
        output_cost = (output_tokens_per_task / 1_000_000) * r.output_price
        cost_per_task = input_cost + output_cost
        
        # 综合智能得分（取AA Intelligence Index）
        intelligence = r.scores.get("AA_Intelligence_Index", 0)
        
        # 效率比 = 智能得分 / 成本
        efficiency_ratio = intelligence / cost_per_task if cost_per_task > 0 else 0
        
        efficiencies.append(CostEfficiency(
            model=r.model,
            cost_per_task=cost_per_task,
            intelligence_score=intelligence,
            efficiency_ratio=efficiency_ratio,
        ))
    
    return efficiencies


def calculate_agentic_efficiency(
    results: List[BenchmarkResult],
) -> Dict[str, float]:
    """
    计算Agent任务效率得分
    综合多个Agent相关基准测试的加权评分
    """
    weights = {
        "GDPVal_AA_v2_Elo": 0.30,
        "AA_Briefcase_Elo": 0.25,
        "CursorBench_v3_2": 0.20,
        "DeepSWE_v1_1": 0.15,
        "Harvey_LAB": 0.10,
    }
    
    agentic_scores = {}
    for r in results:
        weighted_score = 0.0
        for benchmark, weight in weights.items():
            if benchmark in r.scores:
                weighted_score += r.scores[benchmark] * weight
        agentic_scores[r.model] = weighted_score
    
    return agentic_scores


def print_analysis():
    """打印完整分析报告"""
    results = load_benchmark_data()
    
    print("=" * 75)
    print("Grok 4.6 基准测试深度分析")
    print("=" * 75)
    
    # 1. 成本效益分析
    print("\n📊 1. 成本效益分析 (每任务500K输入 + 50K输出)")
    print("-" * 60)
    efficiencies = calculate_cost_efficiency(results)
    for eff in sorted(efficiencies, key=lambda x: x.efficiency_ratio, reverse=True):
        print(f"  {eff.model:25s}")
        print(f"    单任务成本: ${eff.cost_per_task:.2f}")
        print(f"    智能得分:   {eff.intelligence_score:.1f}")
        print(f"    效率比:     {eff.efficiency_ratio:.2f} 分/美元")
        print()
    
    # 2. Agent任务效率
    print("📊 2. Agent任务综合效率")
    print("-" * 60)
    agentic_scores = calculate_agentic_efficiency(results)
    for model, score in sorted(agentic_scores.items(), key=lambda x: x[1], reverse=True):
        print(f"  {model:25s}: {score:.1f}")
    
    # 3. 标准化对比
    print("\n📊 3. Grok 4.6 vs 竞品 (标准化为Grok 4.6=100)")
    print("-" * 60)
    grok46 = results[0]
    for competitor in results[1:]:
        print(f"\n  {competitor.model} 对比 {grok46.model}:")
        for benchmark in grok46.scores:
            grok_score = grok46.scores[benchmark]
            comp_score = competitor.scores.get(benchmark, 0)
            if grok_score > 0:
                ratio = (comp_score / grok_score) * 100
                symbol = "▲" if comp_score > grok_score else "▼" if comp_score < grok_score else "="
                print(f"    {benchmark:25s}: {symbol} {ratio:.0f}%")
    
    # 4. 性价比结论
    print("\n📊 4. 性价比排名")
    print("-" * 60)
    sorted_by_ratio = sorted(efficiencies, key=lambda x: x.efficiency_ratio, reverse=True)
    for i, eff in enumerate(sorted_by_ratio, 1):
        price_ratio = eff.cost_per_task / sorted_by_ratio[-1].cost_per_task
        print(f"  #{i} {eff.model:25s} 成本${eff.cost_per_task:.2f} "
              f"(x{price_ratio:.1f}) 效率比{eff.efficiency_ratio:.1f}")


if __name__ == "__main__":
    print_analysis()

# grok_bot_architecture.py
"""
Grok Bot 持久云电脑架构的简化模拟
"""

import asyncio
import os
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Callable, Dict, List, Optional


class BotRole(Enum):
    SALES_OUTBOUND = "Sales Outbound"
    TALENT_SCOUT = "Talent Scout"
    PAID_MEDIA = "Paid Media"
    EXPENSE_MANAGER = "Expense Manager"
    PRODUCT_PERFORMANCE = "Product Performance"
    BUG_REPRODUCTION = "Bug Reproduction"
    ACCOUNT_HEALTH = "Account Health"
    CHIEF_OF_STAFF = "Chief of Staff"


@dataclass
class CloudComputer:
    """
    持久云电脑环境
    每个用户账户拥有一个独立的VM实例
    """
    account_id: str
    home_dir: str = "/home/grok"
    browser_sessions: Dict[str, Any] = field(default_factory=dict)
    filesystem: Dict[str, str] = field(default_factory=dict)
    environment: Dict[str, str] = field(default_factory=dict)
    is_running: bool = True

    def __post_init__(self):
        self.environment.update({
            "SHELL": "/bin/bash",
            "HOME": self.home_dir,
            "USER": "grok",
        })
        # 初始化文件系统
        self.filesystem = {
            f"{self.home_dir}/.bashrc": "# Grok Bot shell config",
            f"{self.home_dir}/credentials": "",  # 加密存储的凭证
            f"{self.home_dir}/.profile": "# Grok Bot profile",
        }


@dataclass
class GrokBot:
    """
    具名AI代理
    每个Bot在云电脑上拥有自己的会话
    """
    bot_id: str
    name: str
    role: BotRole
    cloud_pc: CloudComputer
    memory: Dict[str, Any] = field(default_factory=dict)
    learned_workflows: Dict[str, Any] = field(default_factory=dict)
    active_tasks: List[str] = field(default_factory=list)

    async def assign_task(self, task: str) -> str:
        """分配任务给Bot"""
        self.active_tasks.append(task)
        print(f"[{self.name}] 接受任务: {task[:50]}...")
        
        # 检查是否有已学习的工作流
        workflow = self._find_matching_workflow(task)
        if workflow:
            print(f"[{self.name}] 找到匹配的工作流，开始执行...")
            return await self._execute_workflow(workflow, task)
        
        # 否则自主执行
        return await self._execute_autonomous(task)
    
    async def learn_workflow(self, name: str, steps: List[Dict]) -> None:
        """
        通过观察学习工作流
        Grok Bot的核心特性：录制用户操作并保存为可重复执行的例程
        """
        self.learned_workflows[name] = {
            "steps": steps,
            "created_at": "2026-08-12",
            "version": 1,
        }
        print(f"[{self.name}] 学习了新工作流: {name} ({len(steps)} steps)")
    
    def _find_matching_workflow(self, task: str) -> Optional[Dict]:
        """查找匹配的已学习工作流"""
        for name, workflow in self.learned_workflows.items():
            if name.lower() in task.lower():
                return workflow
        return None
    
    async def _execute_workflow(self, workflow: Dict, task: str) -> str:
        """执行已学习的工作流"""
        results = []
        for i, step in enumerate(workflow["steps"]):
            action = step.get("action", "unknown")
            print(f"  [{self.name}] 执行步骤 {i+1}: {action}")
            # 模拟执行
            await asyncio.sleep(0.3)
            results.append(f"步骤 {i+1} 完成: {action}")
        return "\n".join(results)
    
    async def _execute_autonomous(self, task: str) -> str:
        """
        自主执行任务
        使用Grok 4.6的推理能力进行规划、执行和验证
        """
        print(f"  [{self.name}] 自主规划执行...")
        # 模拟Grok 4.6的规划-执行-验证循环
        plan = [
            "分析任务需求",
            "确定执行策略",
            "调用工具执行",
            "验证输出结果",
            "生成最终报告",
        ]
        for step in plan:
            print(f"  [{self.name}] ▶ {step}")
            await asyncio.sleep(0.2)
        return f"任务完成: {task}"


@dataclass
class GrokBotOrchestrator:
    """
    Bot编排器
    管理多个Bot的协作和资源分配
    """
    account_id: str
    cloud_pc: CloudComputer
    bots: Dict[str, GrokBot] = field(default_factory=dict)
    
    def create_bot(self, bot_id: str, name: str, role: BotRole) -> GrokBot:
        """创建新的Bot"""
        bot = GrokBot(
            bot_id=bot_id,
            name=name,
            role=role,
            cloud_pc=self.cloud_pc,
        )
        self.bots[bot_id] = bot
        print(f"[Orchestrator] 创建Bot: {name} ({role.value})")
        return bot
    
    async def delegate_task(self, bot_id: str, task: str) -> str:
        """将任务委派给指定的Bot"""
        if bot_id not in self.bots:
            raise ValueError(f"Bot {bot_id} 不存在")
        return await self.bots[bot_id].assign_task(task)
    
    async def multi_bot_collaboration(self, tasks: Dict[str, str]) -> Dict[str, str]:
        """
        多Bot并行协作
        模拟Grok Bot的多Agent并行工作模式
        """
        results = {}
        async def run_bot(bot_id: str, task: str):
            results[bot_id] = await self.delegate_task(bot_id, task)
        
        # 并行执行
        await asyncio.gather(*[
            run_bot(bot_id, task)
            for bot_id, task in tasks.items()
        ])
        return results
    
    def share_context_between_bots(self, source_bot: str, target_bot: str, context: Dict):
        """Bot之间共享上下文信息"""
        if source_bot in self.bots and target_bot in self.bots:
            self.bots[target_bot].memory.update(
                self.bots[source_bot].memory
            )
            print(f"[Orchestrator] 上下文从 {source_bot} 共享到 {target_bot}")


async def main():
    """演示Grok Bot的多Bot协作场景"""
    print("=" * 70)
    print("Grok Bot 持久云电脑架构演示")
    print("=" * 70)
    
    # 创建云电脑
    cloud_pc = CloudComputer(account_id="user-001")
    print(f"\n✅ 创建持久云电脑: {cloud_pc.account_id}")
    
    # 创建编排器
    orchestrator = GrokBotOrchestrator(
        account_id="user-001",
        cloud_pc=cloud_pc,
    )
    
    # 创建多个Bot
    sales_bot = orchestrator.create_bot(
        "bot-001", "销售专员", BotRole.SALES_OUTBOUND
    )
    bug_bot = orchestrator.create_bot(
        "bot-002", "Bug猎人", BotRole.BUG_REPRODUCTION
    )
    staff_bot = orchestrator.create_bot(
        "bot-003", "参谋长", BotRole.CHIEF_OF_STAFF
    )
    
    # 工作流学习演示
    print("\n📚 演示: Bot学习工作流")
    print("-" * 40)
    await sales_bot.learn_workflow("客户线索挖掘", [
        {"action": "打开LinkedIn Sales Navigator"},
        {"action": "搜索目标客户"},
        {"action": "分析客户意向"},
        {"action": "生成个性化邮件"},
        {"action": "留下审核列表"},
    ])
    
    # 多Bot并行协作
    print("\n🤝 演示: 多Bot并行协作")
    print("-" * 40)
    tasks = {
        "bot-001": "挖掘本周的潜在客户线索",
        "bot-002": "复现生产环境中的登录崩溃问题",
        "bot-003": "扫描Slack和邮件，生成今日工作摘要",
    }
    
    results = await orchestrator.multi_bot_collaboration(tasks)
    
    print("\n📋 协作结果:")
    for bot_id, result in results.items():
        bot_name = orchestrator.bots[bot_id].name
        print(f"  [{bot_name}] {result[:50]}...")
    
    print("\n✅ 演示完成")
    print("  核心要点: Bot共享云电脑环境")
    print("  可以在用户关闭设备后继续工作")
    print("  支持多Bot并行协作和上下文共享")


if __name__ == "__main__":
    asyncio.run(main())

# grok46_api_demo.py
"""
Grok 4.6 API 调用完整示例
包含流式推理、多轮对话、工具调用
"""

import json
import time
from typing import AsyncGenerator, Callable, Dict, List, Optional, Any


class Grok46Client:
    """
    Grok 4.6 API 客户端
    支持流式推理、多推理强度、工具调用
    """
    
    def __init__(
        self,
        api_key: str,
        model: str = "grok-4.6",
        base_url: str = "https://api.x.ai/v1",
    ):
        self.api_key = api_key
        self.model = model
        self.base_url = base_url
        self.conversation_history: List[Dict] = []
    
    def set_reasoning_effort(self, effort: str):
        """
        设置推理强度
        可选值: low | medium | high | xhigh
        
        - low: 快速响应，适合简单任务
        - medium: 平衡模式
        - high: 深度推理，适合复杂任务
        - xhigh: 极致推理，适合最困难的问题
        """
        valid_efforts = ["low", "medium", "high", "xhigh"]
        if effort not in valid_efforts:
            raise ValueError(f"推理强度必须为: {valid_efforts}")
        self.reasoning_effort = effort
    
    async def chat_completion(
        self,
        messages: List[Dict],
        temperature: float = 0.7,
        max_tokens: int = 16384,
        stream: bool = False,
        tools: Optional[List[Dict]] = None,
    ) -> Dict:
        """
        聊天补全
        支持工具调用和流式输出
        """
        request_body = {
            "model": self.model,
            "messages": messages,
            "temperature": temperature,
            "max_tokens": max_tokens,
            "stream": stream,
        }
        
        if hasattr(self, 'reasoning_effort'):
            request_body["reasoning_effort"] = self.reasoning_effort
        
        if tools:
            request_body["tools"] = tools
            request_body["tool_choice"] = "auto"
        
        # 模拟API调用
        print(f"  [Grok 4.6] 推理强度: {getattr(self, 'reasoning_effort', 'medium')}")
        print(f"  [Grok 4.6] 输入消息数: {len(messages)}")
        
        # 模拟响应
        response = {
            "id": f"chatcmpl-{int(time.time())}",
            "object": "chat.completion",
            "created": int(time.time()),
            "model": self.model,
            "choices": [{
                "index": 0,
                "message": {
                    "role": "assistant",
                    "content": "模拟的Grok 4.6响应内容",
                },
                "finish_reason": "stop",
            }],
            "usage": {
                "prompt_tokens": 500,
                "completion_tokens": 200,
                "total_tokens": 700,
            },
        }
        
        return response
    
    def add_tool(self, name: str, description: str, parameters: Dict):
        """注册工具供模型调用"""
        tool = {
            "type": "function",
            "function": {
                "name": name,
                "description": description,
                "parameters": parameters,
            },
        }
        if not hasattr(self, '_tools'):
            self._tools = []
        self._tools.append(tool)
    
    async def agentic_loop(
        self,
        task: str,
        max_iterations: int = 10,
        tool_handlers: Optional[Dict[str, Callable]] = None,
    ) -> str:
        """
        长程Agent循环
        模型自主规划、调用工具、验证结果
        
        参数:
            task: 任务描述
            max_iterations: 最大迭代次数
            tool_handlers: 工具名称到处理函数的映射
        """
        messages = [
            {
                "role": "system",
                "content": (
                    "你是Grok 4.6，一个专注于长程Agent任务的AI助手。"
                    "你可以使用工具来完成复杂任务。"
                    "在每个步骤中，你应该：\n"
                    "1. 分析当前状态\n"
                    "2. 决定下一步行动\n"
                    "3. 调用工具或生成中间结果\n"
                    "4. 验证输出质量\n"
                    "5. 必要时自我修正"
                ),
            },
            {"role": "user", "content": task},
        ]
        
        print(f"\n{'='*60}")
        print(f"Grok 4.6 Agentic Loop 开始")
        print(f"任务: {task[:80]}...")
        print(f"最大迭代: {max_iterations}")
        print(f"{'='*60}")
        
        for iteration in range(max_iterations):
            print(f"\n[迭代 {iteration + 1}/{max_iterations}]")
            
            response = await self.chat_completion(
                messages=messages,
                tools=getattr(self, '_tools', None),
            )
            
            assistant_message = response["choices"][0]["message"]
            messages.append(assistant_message)
            
            # 检查是否有工具调用
            if "tool_calls" in assistant_message:
                for tool_call in assistant_message["tool_calls"]:
                    tool_name = tool_call["function"]["name"]
                    tool_args = json.loads(tool_call["function"]["arguments"])
                    
                    print(f"  🔧 调用工具: {tool_name}({tool_args})")
                    
                    # 执行工具
                    if tool_handlers and tool_name in tool_handlers:
                        result = tool_handlers[tool_name](**tool_args)
                    else:
                        result = f"模拟工具 {tool_name} 执行结果"
                    
                    # 添加工具结果
                    messages.append({
                        "role": "tool",
                        "tool_call_id": tool_call["id"],
                        "content": json.dumps(result),
                    })
                    print(f"  ✅ 工具结果: {str(result)[:80]}...")
            else:
                # 没有工具调用，检查是否任务完成
                content = assistant_message.get("content", "")
                if "任务完成" in content or "FINAL" in content.upper():
                    print(f"\n  ✅ 任务完成!")
                    return content
                
                print(f"  💬 模型回复: {content[:80]}...")
        
        # 达到最大迭代次数
        final = messages[-1]["content"] if messages else "未生成结果"
        print(f"\n  ⚠️ 达到最大迭代次数 ({max_iterations})")
        return final


# 实用工具函数
def create_search_tool() -> Dict:
    """创建搜索工具定义"""
    return {
        "type": "function",
        "function": {
            "name": "web_search",
            "description": "搜索互联网上的信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "搜索关键词",
                    },
                    "max_results": {
                        "type": "integer",
                        "description": "最大返回结果数",
                        "default": 5,
                    },
                },
                "required": ["query"],
            },
        },
    }


def create_code_executor_tool() -> Dict:
    """创建代码执行工具定义"""
    return {
        "type": "function",
        "function": {
            "name": "execute_python",
            "description": "执行Python代码并返回结果",
            "parameters": {
                "type": "object",
                "properties": {
                    "code": {
                        "type": "string",
                        "description": "要执行的Python代码",
                    },
                    "timeout": {
                        "type": "integer",
                        "description": "超时时间（秒）",
                        "default": 30,
                    },
                },
                "required": ["code"],
            },
        },
    }


async def demo_long_running_task():
    """
    演示Grok 4.6处理长程任务的能力
    场景：从零构建一个数据分析应用
    """
    client = Grok46Client(api_key="demo-key")
    client.set_reasoning_effort("high")
    
    # 注册工具
    client.add_tool("web_search", "搜索互联网信息", {
        "type": "object",
        "properties": {
            "query": {"type": "string", "description": "搜索关键词"},
        },
        "required": ["query"],
    })
    client.add_tool("execute_python", "执行Python代码", {
        "type": "object",
        "properties": {
            "code": {"type": "string", "description": "Python代码"},
        },
        "required": ["code"],
    })
    
    # 定义工具处理函数
    tool_handlers = {
        "web_search": lambda query: {"results": [f"关于{query}的搜索结果"]},
        "execute_python": lambda code: {"output": "代码执行结果", "success": True},
    }
    
    # 执行长程任务
    task = """
    我需要一个数据分析仪表盘应用。
    请完成以下步骤：
    1. 研究当前流行的数据可视化框架
    2. 设计应用架构
    3. 实现核心数据加载和可视化功能
    4. 创建交互式仪表盘界面
    5. 编写测试用例
    6. 生成部署说明
    """
    
    result = await client.agentic_loop(
        task=task,
        max_iterations=15,
        tool_handlers=tool_handlers,
    )
    
    print(f"\n最终结果:\n{result}")
    return result


if __name__ == "__main__":
    import asyncio
    asyncio.run(demo_long_running_task())