GPT-5.6 Kindle-Alpha 深度技术解析：150万Token、全双工语音与Agentic工作流的暴力美学

"""
GPT-5.6 分片层级注意力机制的核心实现（原型代码）
基于泄露的架构推理重构，非官方实现
"""

import torch
import torch.nn as nn
import torch.nn.functional as F
import math
from dataclasses import dataclass
from typing import Optional, Tuple, List

@dataclass
class SHAConfig:
    hidden_size: int = 7168
    num_attention_heads: int = 64
    num_kv_heads: int = 8  # MLA: 8 KV heads, 64 query heads
    segment_size: int = 8192  # 每个分片8K tokens
    num_global_tokens: int = 1024  # 全局摘要token数
    sliding_window: int = 4096  # 滑动窗口局部注意力
    max_seq_len: int = 1_500_000
    rope_theta: float = 1_000_000.0

class SegmentedHierarchicalAttention(nn.Module):
    """
    分片层级注意力（SHA）：
    1) 将序列分片，每片内做完整自注意力
    2) 提取分片摘要token，在摘要层做全局交叉注意力
    3) 滑动窗口捕捉局部依赖
    """
    def __init__(self, config: SHAConfig):
        super().__init__()
        self.config = config
        self.hidden_size = config.hidden_size
        self.num_heads = config.num_attention_heads
        self.num_kv_heads = config.num_kv_heads
        self.head_dim = config.hidden_size // config.num_attention_heads
        
        # 投影矩阵 - 使用MLA (Multi-head Latent Attention) 压缩
        self.q_proj = nn.Linear(config.hidden_size, config.hidden_size, bias=False)
        self.kv_proj = nn.Linear(config.hidden_size, 
                                  config.num_kv_heads * self.head_dim * 2, bias=False)
        self.o_proj = nn.Linear(config.hidden_size, config.hidden_size, bias=False)
        
        # 分片摘要提取器
        self.segment_summarizer = nn.Sequential(
            nn.Linear(config.hidden_size, config.hidden_size * 2),
            nn.GELU(),
            nn.Linear(config.hidden_size * 2, config.hidden_size),
        )
        
        # 全局摘要token（可学习的queried tokens）
        self.global_summary_tokens = nn.Parameter(
            torch.randn(1, config.num_global_tokens, config.hidden_size)
        )
        
        # 滑动窗口掩码
        self.register_buffer("sliding_mask", self._build_sliding_mask(
            config.segment_size, config.sliding_window
        ))
        
        # RoPE位置编码（扩展至150万）
        self.rope_cache = self._build_rope_cache(config.max_seq_len, config.rope_theta)
        
    def _build_sliding_mask(self, seg_size: int, window: int):
        """构建滑动窗口注意力掩码"""
        mask = torch.full((seg_size, seg_size), float('-inf'))
        for i in range(seg_size):
            start = max(0, i - window)
            mask[i, start:i+1] = 0.0
        return mask
    
    def _build_rope_cache(self, max_len: int, theta: float):
        """构建支持150万token的RoPE位置缓存"""
        position = torch.arange(max_len, dtype=torch.float32)
        dim = self.head_dim
        freqs = 1.0 / (theta ** (torch.arange(0, dim, 2).float() / dim))
        # 外积: [max_len, dim/2] -> [max_len, dim]
        angles = position[:, None] * freqs[None, :]
        angles = torch.cat([angles, angles], dim=-1)
        return angles  # [max_len, head_dim]
    
    def _apply_rope(self, x: torch.Tensor, positions: torch.Tensor):
        """应用旋转位置编码"""
        cos = torch.cos(self.rope_cache[positions])
        sin = torch.sin(self.rope_cache[positions])
        # 旋转操作: x * cos + rotate_half(x) * sin
        x_rot = torch.stack([-x[..., 1::2], x[..., ::2]], dim=-1).reshape(x.shape)
        return x * cos + x_rot * sin
    
    def segment_attention(self, 
                          hidden_states: torch.Tensor,
                          segment_ids: torch.Tensor) -> torch.Tensor:
        """
        分片内注意力计算
        hidden_states: [batch, seq_len, hidden]
        segment_ids: [batch, seq_len] 每个token所属分片ID
        """
        batch, seq_len, _ = hidden_states.shape
        num_segments = segment_ids.max().item() + 1
        
        outputs = torch.zeros_like(hidden_states)
        
        for seg_id in range(num_segments):
            # 提取当前分片的token
            seg_mask = (segment_ids == seg_id)
            seg_indices = seg_mask.nonzero(as_tuple=True)[1]  # 仅batch=0简化
            seg_hidden = hidden_states[:, seg_indices, :]
            seg_len = seg_hidden.size(1)
            
            if seg_len == 0:
                continue
            
            # 投影到Q/K/V
            q = self.q_proj(seg_hidden)
            k = self.kv_proj(seg_hidden)[:, :, :self.num_kv_heads * self.head_dim]
            v = self.kv_proj(seg_hidden)[:, :, self.num_kv_heads * self.head_dim:]
            
            # 重塑多头
            q = q.view(batch, seg_len, self.num_heads, self.head_dim).transpose(1, 2)
            k = k.view(batch, seg_len, self.num_kv_heads, self.head_dim).transpose(1, 2)
            v = v.view(batch, seg_len, self.num_kv_heads, self.head_dim).transpose(1, 2)
            
            # RoPE
            positions = seg_indices.to(hidden_states.device)
            q = self._apply_rope(q, positions)
            k = self._apply_rope(k, positions)
            
            # 滑动窗口注意力
            window_mask = self.sliding_mask[:seg_len, :seg_len].to(hidden_states.device)
            
            # MLA: KV head扩展到所有Q head
            # 此处简化为重复KV heads
            k = k.repeat_interleave(self.num_heads // self.num_kv_heads, dim=1)
            v = v.repeat_interleave(self.num_heads // self.num_kv_heads, dim=1)
            
            attn_weights = torch.matmul(q, k.transpose(-2, -1)) / math.sqrt(self.head_dim)
            attn_weights = attn_weights + window_mask.unsqueeze(0).unsqueeze(0)
            attn_weights = F.softmax(attn_weights, dim=-1)
            
            seg_output = torch.matmul(attn_weights, v)
            seg_output = seg_output.transpose(1, 2).contiguous().view(batch, seg_len, -1)
            seg_output = self.o_proj(seg_output)
            
            outputs[:, seg_indices, :] = seg_output
        
        return outputs
    
    def global_cross_attention(self, 
                                segment_summaries: torch.Tensor,
                                hidden_states: torch.Tensor) -> torch.Tensor:
        """
        全局摘要交叉注意力
        segment_summaries: [batch, num_segments, hidden]
        hidden_states: [batch, seq_len, hidden]
        """
        # 可学习的global query
        global_q = self.global_summary_tokens.expand(
            hidden_states.size(0), -1, -1
        )
        global_q = self.q_proj(global_q)
        
        # segment summaries as KV
        seg_k = self.kv_proj(segment_summaries)[:, :, :self.num_kv_heads * self.head_dim]
        seg_v = self.kv_proj(segment_summaries)[:, :, self.num_kv_heads * self.head_dim:]
        
        # 重塑 + 注意力
        batch, num_seg, _ = segment_summaries.shape
        
        global_q = global_q.view(batch, -1, self.num_heads, self.head_dim).transpose(1, 2)
        seg_k = seg_k.view(batch, num_seg, self.num_kv_heads, self.head_dim).transpose(1, 2)
        seg_v = seg_v.view(batch, num_seg, self.num_kv_heads, self.head_dim).transpose(1, 2)
        
        seg_k = seg_k.repeat_interleave(self.num_heads // self.num_kv_heads, dim=1)
        seg_v = seg_v.repeat_interleave(self.num_heads // self.num_kv_heads, dim=1)
        
        attn = torch.matmul(global_q, seg_k.transpose(-2, -1)) / math.sqrt(self.head_dim)
        attn = F.softmax(attn, dim=-1)
        global_output = torch.matmul(attn, seg_v)
        
        # 将全局信息广播回所有token
        global_output = global_output.transpose(1, 2).contiguous().view(batch, -1, self.hidden_size)
        global_output = self.o_proj(global_output)
        
        # 通过门控融合
        gate = torch.sigmoid(global_output.mean(dim=1, keepdim=True))
        return hidden_states + gate * global_output.unsqueeze(1).expand(-1, hidden_states.size(1), -1)
    
    def forward(self, hidden_states: torch.Tensor, 
                segment_ids: torch.Tensor,
                positions: torch.Tensor) -> torch.Tensor:
        # 1) 分片内注意力
        local_out = self.segment_attention(hidden_states, segment_ids)
        
        # 2) 提取分片摘要
        num_segments = segment_ids.max().item() + 1
        summaries = []
        for seg_id in range(num_segments):
            seg_hidden = local_out[:, segment_ids[0] == seg_id, :]
            if seg_hidden.size(1) > 0:
                summary = self.segment_summarizer(seg_hidden.mean(dim=1, keepdim=True))
                summaries.append(summary)
        
        if summaries:
            seg_summaries = torch.cat(summaries, dim=1)
            # 3) 全局交叉注意力
            output = self.global_cross_attention(seg_summaries, local_out)
        else:
            output = local_out
        
        return output

"""
GPT-Bidi-1 全双工语音引擎核心架构（原型重构）
基于泄露的逆向工程推理，非官方实现
"""

import asyncio
import numpy as np
from dataclasses import dataclass, field
from typing import Optional, AsyncGenerator, List, Callable
import torch
import torch.nn as nn
import torch.nn.functional as F

@dataclass
class BidiConfig:
    sample_rate: int = 24000
    frame_ms: int = 20  # 20ms帧
    hop_length: int = 480  # 24000 * 0.02
    num_mel_bins: int = 128
    encoder_dim: int = 1024
    num_encoder_layers: int = 6
    num_decoder_layers: int = 4
    num_heads: int = 16
    
    # 全双工特定参数
    overlap_buffer_ms: int = 500  # 500ms重叠缓冲区
    interruption_threshold: float = 0.65  # 打断检测阈值
    replan_latency_ms: int = 80  # 重规划延迟

class StreamingAudioEncoder(nn.Module):
    """
    流式音频编码器：逐帧处理，保持状态
    """
    def __init__(self, config: BidiConfig):
        super().__init__()
        self.config = config
        
        # Mel频谱提取
        self.mel_conv = nn.Sequential(
            nn.Conv1d(config.num_mel_bins, config.encoder_dim, 
                      kernel_size=3, padding=1),
            nn.GELU(),
            nn.Conv1d(config.encoder_dim, config.encoder_dim, 
                      kernel_size=3, padding=1, stride=2),
        )
        
        # 因果Transformer编码器（只看到过去）
        self.causal_encoder = nn.TransformerEncoder(
            nn.TransformerEncoderLayer(
                d_model=config.encoder_dim,
                nhead=config.num_heads,
                dim_feedforward=config.encoder_dim * 4,
                batch_first=True,
            ),
            num_layers=config.num_encoder_layers,
        )
        
        # 因果掩码
        self.register_buffer("causal_mask", 
            torch.triu(torch.full((4096, 4096), float('-inf')), diagonal=1))
        
        # 音频语义投影
        self.semantic_proj = nn.Linear(config.encoder_dim, config.encoder_dim)
        
    def forward(self, mel_frames: torch.Tensor, 
                state: Optional[torch.Tensor] = None):
        """
        mel_frames: [batch, n_mels, T] 当前帧批
        state: 上一个编码状态
        """
        x = self.mel_conv(mel_frames)  # [batch, dim, T/2]
        x = x.transpose(1, 2)  # [batch, T/2, dim]
        
        # 因果编码（保持流式性质）
        seq_len = x.size(1)
        mask = self.causal_mask[:seq_len, :seq_len].to(x.device)
        x = self.causal_encoder(x, mask=mask, is_causal=True)
        
        return self.semantic_proj(x)

class InterruptionDetector(nn.Module):
    """
    打断检测器：实时检测用户是否在AI说话时插话
    """
    def __init__(self, config: BidiConfig):
        super().__init__()
        self.config = config
        
        self.audio_context_encoder = nn.LSTM(
            input_size=config.encoder_dim,
            hidden_size=256,
            num_layers=2,
            batch_first=True,
            bidirectional=False,
        )
        
        self.interruption_classifier = nn.Sequential(
            nn.Linear(256, 128),
            nn.ReLU(),
            nn.Linear(128, 3),  # 0: 用户静音, 1: 用户倾听, 2: 用户打断
        )
        
        # 能量门控
        self.energy_threshold = config.interruption_threshold
        
    def forward(self, 
                audio_embedding: torch.Tensor,
                user_energy: torch.Tensor) -> torch.Tensor:
        """
        返回: [batch, T, 3] 打断概率分布
        """
        # LSTM编码音频上下文
        context, _ = self.audio_context_encoder(audio_embedding)
        
        # 分类
        logits = self.interruption_classifier(context)
        
        # 能量门控：低能量时强制为静音
        energy_mask = (user_energy < self.energy_threshold).unsqueeze(-1).float()
        logits[:, :, 0] += energy_mask.squeeze(-1) * 10.0  # 增强静音概率
        
        return F.softmax(logits, dim=-1)

class DualStreamProcessor:
    """
    双流处理器：同时处理输入(用户)和输出(AI)音频流
    """
    def __init__(self, config: BidiConfig):
        self.config = config
        
        # 两个独立的音频编码器
        self.input_encoder = StreamingAudioEncoder(config)
        self.output_encoder = StreamingAudioEncoder(config)
        
        # 共享的打断检测器
        self.interruption_detector = InterruptionDetector(config)
        
        # 语义融合器
        self.semantic_fuser = nn.TransformerDecoder(
            nn.TransformerDecoderLayer(
                d_model=config.encoder_dim,
                nhead=config.num_heads,
                dim_feedforward=config.encoder_dim * 4,
                batch_first=True,
            ),
            num_layers=config.num_decoder_layers,
        )
        
        # 输出音频生成器
        self.audio_decoder = nn.Sequential(
            nn.ConvTranspose1d(config.encoder_dim, config.num_mel_bins,
                               kernel_size=4, stride=2, padding=1),
            nn.GELU(),
            nn.Conv1d(config.num_mel_bins, config.num_mel_bins,
                      kernel_size=3, padding=1),
        )
        
        # 对话状态
        self.dialog_state = {
            "turn": "ai_speaking",  # ai_speaking / user_speaking / interruption
            "pending_response": "",
            "interruption_history": [],
        }
        
    async def process_stream(self, 
                              input_stream: AsyncGenerator[np.ndarray, None],
                              lm_generate: Callable):
        """
        全双工流式处理主循环
        """
        input_buffer = []
        output_buffer = []
        
        async for audio_frame in input_stream:
            # 1) 编码输入帧
            mel = self._audio_to_mel(audio_frame)
            input_emb = self.input_encoder(mel.unsqueeze(0))
            input_buffer.append(input_emb)
            
            # 2) 检测打断
            user_energy = np.abs(audio_frame).mean()
            interruption_probs = self.interruption_detector(
                torch.cat(input_buffer[-20:], dim=1),
                torch.tensor([[user_energy]])
            )
            
            # 3) 判断打断
            is_interruption = (
                interruption_probs[0, -1, 2].item() > 0.5 and
                self.dialog_state["turn"] == "ai_speaking"
            )
            
            if is_interruption:
                # --- 打断处理 ---
                self.dialog_state["turn"] = "interruption"
                
                # 记录打断点
                interrupt_audio = torch.cat(input_buffer[-10:], dim=1)
                self.dialog_state["interruption_history"].append(interrupt_audio)
                
                # 融合打断语义与当前输出上下文
                output_ctx = torch.cat(output_buffer[-10:], dim=1)
                fused = self.semantic_fuser(
                    input_emb,  # 新的用户输入作为query
                    torch.cat([output_ctx], dim=1),  # 之前输出作为memory
                )
                
                # 重规划回应
                new_response = await lm_generate(
                    fused, 
                    mode="replan",
                    interrupt_context=self.dialog_state["interruption_history"][-1]
                )
                
                self.dialog_state["pending_response"] = new_response
                self.dialog_state["turn"] = "ai_speaking"
            
            # 4) 生成输出音频
            if self.dialog_state["turn"] == "ai_speaking":
                output_emb = self.output_encoder(
                    self._text_to_mel(self.dialog_state["pending_response"])
                )
                
                # 双流语义融合（避免AI自己听到自己）
                fused = self.semantic_fuser(
                    output_emb,
                    torch.cat(input_buffer[-5:], dim=1),  # 监听用户
                )
                
                output_audio = self.audio_decoder(fused.transpose(1, 2))
                output_buffer.append(fused)
                
                yield output_audio.squeeze(0).cpu().numpy()
    
    def _audio_to_mel(self, audio: np.ndarray) -> torch.Tensor:
        """音频转Mel频谱"""
        # 简化的Mel提取
        return torch.from_numpy(audio).float().unsqueeze(0)
    
    def _text_to_mel(self, text: str) -> torch.Tensor:
        """文本转Mel频谱（用于TTS）"""
        # 此处应该是调用LLM生成+声码器
        return torch.randn(1, self.config.num_mel_bins, 100)

"""
GPT-5.6 视觉UI代码生成管线（原型重构）
"""

import torch
import torch.nn as nn
import torch.nn.functional as F
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass

@dataclass
class VisionUIConfig:
    image_size: int = 1024
    patch_size: int = 16
    visual_encoder_dim: int = 2048
    num_visual_layers: int = 32
    num_visual_heads: int = 32
    
    ui_tokenizer_vocab: int = 32768  # UI组件token化词汇表
    ui_embed_dim: int = 2048
    
    max_ui_elements: int = 500  # 单页面最大UI元素数

class MultiScaleVisualEncoder(nn.Module):
    """
    多尺度视觉编码器：同时理解整体布局和局部细节
    """
    def __init__(self, config: VisionUIConfig):
        super().__init__()
        self.config = config
        
        # 全局布局理解（低分辨率、大感受野）
        self.global_encoder = nn.Sequential(
            nn.Conv2d(3, 128, kernel_size=7, stride=4, padding=3),
            nn.GELU(),
            nn.Conv2d(128, 256, kernel_size=3, stride=2, padding=1),
            nn.GELU(),
            nn.AdaptiveAvgPool2d((16, 16)),
        )
        
        # 局部细节理解（高分辨率、小感受野）
        self.local_encoder = nn.Sequential(
            nn.Conv2d(3, 64, kernel_size=3, stride=1, padding=1),
            nn.GELU(),
            nn.Conv2d(64, 128, kernel_size=3, stride=2, padding=1),
            nn.GELU(),
        )
        
        # 视觉-语言交叉注意力
        self.cross_attn = nn.MultiheadAttention(
            embed_dim=config.visual_encoder_dim,
            num_heads=config.num_visual_heads,
            batch_first=True,
        )
        
        # 布局编码器（元素空间关系）
        self.layout_encoder = nn.TransformerEncoder(
            nn.TransformerEncoderLayer(
                d_model=config.visual_encoder_dim,
                nhead=16,
                dim_feedforward=config.visual_encoder_dim * 4,
                batch_first=True,
            ),
            num_layers=4,
        )
        
    def forward(self, pixel_values: torch.Tensor) -> torch.Tensor:
        """
        pixel_values: [batch, 3, H, W]
        return: [batch, seq_len, dim] 视觉特征序列
        """
        B = pixel_values.shape[0]
        
        # 全局布局特征
        global_feat = self.global_encoder(pixel_values)  # [B, 256, 16, 16]
        global_feat = global_feat.flatten(2).transpose(1, 2)  # [B, 256, dim]
        
        # 局部细节特征
        local_feat = self.local_encoder(pixel_values)  # [B, 128, H/2, W/2]
        local_feat = local_feat.flatten(2).transpose(1, 2)
        
        # 投影到统一维度
        global_proj = nn.Linear(256, self.config.visual_encoder_dim).to(pixel_values.device)
        local_proj = nn.Linear(128, self.config.visual_encoder_dim).to(pixel_values.device)
        
        global_feat = global_proj(global_feat)
        local_feat = local_proj(local_feat)
        
        # 融合多尺度特征
        visual_features = torch.cat([global_feat, local_feat], dim=1)
        
        # 布局关系建模
        visual_features = self.layout_encoder(visual_features)
        
        return visual_features

class UIComponentDecoder(nn.Module):
    """
    UI组件解码器：将视觉特征转换为结构化的UI代码
    """
    def __init__(self, config: VisionUIConfig):
        super().__init__()
        self.config = config
        
        # UI组件词汇表
        self.component_embed = nn.Embedding(
            config.ui_tokenizer_vocab, config.ui_embed_dim
        )
        
        # 视觉到UI的交叉注意力解码器
        self.ui_decoder = nn.TransformerDecoder(
            nn.TransformerDecoderLayer(
                d_model=config.ui_embed_dim,
                nhead=16,
                dim_feedforward=config.ui_embed_dim * 4,
                batch_first=True,
            ),
            num_layers=8,
        )
        
        # UI布局预测头
        self.layout_head = nn.Sequential(
            nn.Linear(config.ui_embed_dim, 512),
            nn.GELU(),
            nn.Linear(512, 4),  # [x, y, w, h] 归一化坐标
        )
        
        # 样式属性预测头
        self.style_head = nn.Sequential(
            nn.Linear(config.ui_embed_dim, 512),
            nn.GELU(),
            nn.Linear(512, 128),  # 样式embedding
        )
        
        # UI组件类型分类
        self.type_head = nn.Linear(config.ui_embed_dim, 50)  # 50种UI组件类型
        
    def forward(self, 
                visual_features: torch.Tensor,
                partial_ui: Optional[torch.Tensor] = None) -> Dict[str, torch.Tensor]:
        """
        从视觉特征生成UI组件序列
        """
        if partial_ui is not None:
            # 自回归生成，逐步预测
            ui_embed = self.component_embed(partial_ui)
            ui_features = self.ui_decoder(ui_embed, visual_features)
        else:
            # 首次调用，使用零嵌入
            batch_size = visual_features.size(0)
            start_token = torch.zeros(batch_size, 1, dtype=torch.long, 
                                      device=visual_features.device)
            ui_embed = self.component_embed(start_token)
            ui_features = self.ui_decoder(ui_embed, visual_features)
        
        return {
            "layout": self.layout_head(ui_features),  # 位置
            "style": self.style_head(ui_features),     # 样式
            "type": self.type_head(ui_features),       # 组件类型
        }

"""
GPT-5.6 Agentic工作流编排框架（原型重构）
"""

import asyncio
from enum import Enum
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Callable, Awaitable
import json
import time

class TaskStatus(Enum):
    PENDING = "pending"
    RUNNING = "running"
    SUCCESS = "success"
    FAILED = "failed"
    RETRYING = "retrying"

@dataclass
class AgentTask:
    id: str
    objective: str
    subtasks: list = field(default_factory=list)
    context: Dict[str, Any] = field(default_factory=dict)
    tools: List[str] = field(default_factory=list)
    status: TaskStatus = TaskStatus.PENDING
    result: Any = None
    error_count: int = 0
    max_retries: int = 3
    created_at: float = 0.0
    
    def to_prompt(self) -> str:
        return f"""TASK ID: {self.id}
OBJECTIVE: {self.objective}
CONTEXT: {json.dumps(self.context, ensure_ascii=False)}
AVAILABLE TOOLS: {', '.join(self.tools)}
PREVIOUS ATTEMPTS: {self.error_count}
STATUS: {self.status.value}"""

class AgenticWorkflowEngine:
    """
    长时间自主Agent工作流引擎
    特点：自我纠错、动态重规划、上下文管理
    """
    def __init__(self, model_generate: Callable):
        self.model_generate = model_generate
        self.tasks: Dict[str, AgentTask] = {}
        self.completed: List[str] = []
        self.sandbox: Dict[str, Any] = {}  # 共享沙箱
        
    async def execute_workflow(self, main_objective: str) -> Dict[str, Any]:
        """执行完整工作流"""
        start_time = time.time()
        
        # 1) 任务分解
        subtasks = await self._decompose(main_objective)
        
        # 2) 执行任务DAG
        execution_plan = self._build_execution_plan(subtasks)
        
        results = {}
        for stage in execution_plan:
            stage_results = await asyncio.gather(*[
                self._execute_with_retry(task_id)
                for task_id in stage
            ])
            for task_id, result in zip(stage, stage_results):
                results[task_id] = result
        
        elapsed = time.time() - start_time
        
        return {
            "objective": main_objective,
            "total_tasks": len(subtasks),
            "elapsed_hours": elapsed / 3600,
            "task_results": results,
            "iteration_count": len(execution_plan),
        }
    
    async def _decompose(self, objective: str) -> List[AgentTask]:
        """将主目标分解为可执行的任务DAG"""
        prompt = f"""Analyze this objective and decompose it into executable subtasks:
Objective: {objective}

For each subtask, specify:
- id: unique identifier
- objective: clear single-task description
- tools needed: list of tool names
- dependencies: which subtask IDs must complete first
- estimated complexity: low/medium/high

Return as JSON array."""
        
        response = await self.model_generate(prompt, max_tokens=4000)
        subtask_specs = json.loads(response)
        
        tasks = []
        for spec in subtask_specs:
            task = AgentTask(
                id=spec["id"],
                objective=spec["objective"],
                tools=spec.get("tools", []),
                context={"dependencies": spec.get("dependencies", [])},
                created_at=time.time(),
            )
            self.tasks[task.id] = task
            tasks.append(task)
        
        return tasks
    
    def _build_execution_plan(self, tasks: List[AgentTask]) -> List[List[str]]:
        """构建并行执行计划（dag scheduling）"""
        # 拓扑排序
        in_degree = {t.id: len(t.context.get("dependencies", [])) for t in tasks}
        adj = {t.id: [] for t in tasks}
        for t in tasks:
            for dep in t.context.get("dependencies", []):
                adj[dep].append(t.id)
        
        plan = []
        queue = [t.id for t in tasks if in_degree[t.id] == 0]
        
        while queue:
            plan.append(list(queue))
            new_queue = []
            for task_id in queue:
                for next_id in adj[task_id]:
                    in_degree[next_id] -= 1
                    if in_degree[next_id] == 0:
                        new_queue.append(next_id)
            queue = new_queue
        
        return plan
    
    async def _execute_with_retry(self, task_id: str) -> Any:
        """带重试的任务执行"""
        task = self.tasks[task_id]
        task.status = TaskStatus.RUNNING
        
        while task.error_count < task.max_retries:
            try:
                # 注入依赖结果作为上下文
                dep_context = {}
                for dep_id in task.context.get("dependencies", []):
                    if dep_id in self.completed:
                        dep_context[dep_id] = self.tasks[dep_id].result
                task.context["dependency_results"] = dep_context
                
                # 执行任务
                prompt = task.to_prompt() + "\n\nExecute this task now."
                result = await self.model_generate(
                    prompt, 
                    max_tokens=16000,
                    tools=task.tools,
                )
                
                # 自我验证
                verification = await self._self_verify(task, result)
                
                if verification["passed"]:
                    task.result = result
                    task.status = TaskStatus.SUCCESS
                    self.completed.append(task_id)
                    return result
                else:
                    task.error_count += 1
                    task.status = TaskStatus.RETRYING
                    task.context["last_error"] = verification["feedback"]
                    task.context["verification_feedback"] = verification["feedback"]
                    
                    # 动态调整策略
                    await self._adapt_strategy(task, verification)
                    
            except Exception as e:
                task.error_count += 1
                task.status = TaskStatus.RETRYING
                task.context["last_error"] = str(e)
        
        task.status = TaskStatus.FAILED
        return {"error": f"Max retries ({task.max_retries}) exceeded"}
    
    async def _self_verify(self, task: AgentTask, result: Any) -> Dict:
        """自我验证：检查输出是否正确"""
        prompt = f"""Verify if the following task was completed correctly:

TASK OBJECTIVE: {task.objective}
RESULT: {json.dumps(result, ensure_ascii=False)[:5000]}

Check:
1. Is the output correct and complete?
2. Are there any obvious errors?
3. Is the output in the expected format?

Response format: {{"passed": true/false, "feedback": "..."}}"""
        
        verification = await self.model_generate(prompt, max_tokens=1000)
        return json.loads(verification)
    
    async def _adapt_strategy(self, task: AgentTask, verification: Dict):
        """根据验证反馈动态调整策略"""
        prompt = f"""The task failed verification. Adapt the execution strategy:

TASK: {task.objective}
VERIFICATION FEEDBACK: {verification.get('feedback', '')}
PREVIOUS ATTEMPTS: {task.error_count}

Suggest a different approach for the next attempt.
Provide specific changes to the execution plan."""
        
        adaptation = await self.model_generate(prompt, max_tokens=2000)
        task.context["adapted_strategy"] = adaptation