KLong — 超长时域 Agent 训练：轨迹分割 SFT + 渐进 RL

Skill-KLong-Long-Horizon-Agent-Training · 16-智能体工程

causalexperimentforecastingoptimizationmulti_agentdata_collectionpricing供应链与补货推荐与搜索数据采集与治理MAS与智能体工程定价与利润风控与合规WF-A 智能补货WF-D 选品扫描WF-F 动态定价WF-G Listing内容优化

实现难度⭐⭐⭐⭐☆

业务优先级⭐⭐⭐⭐☆

业务视角

适用角色CTO / 技术负责人 · 产品经理 · 数据工程师

适用平台跨境运营 AI Agent 工程落地 · Amazon SP API + LLM 集成 · 多平台数据采集 Agent

什么情况下用想把 AI 集成到业务系统，但 LLM 稳定性差、幻觉问题、成本控制都是挑战；Agent 任务失败了不知道哪步出了问题

成功是什么样的AI Agent 在生产环境稳定运行，失败可追踪，成本可控，复杂任务完成率 >85%

业务痛点

LLM 返回结果不稳定不可靠AI 幻觉导致业务决策错误Agent 任务失败了不知道哪步出问题AI 调用成本控制不住

1. 解决的问题

训练 LLM Agent 执行超长时域任务（50+ 步）面临两大瓶颈

2. 核心算法逻辑

训练 LLM Agent 执行超长时域任务（50+ 步）面临两大瓶颈：

3. 业务应用场景

母婴出海选品调研是典型的超长时域任务，完整流程包含 50+ 步骤：

现有方案：用通用 LLM 逐步执行，但： - 超长上下文导致中途"遗忘"前期调研结论 - 每次重新启动都要重发全量背景，成本极高 - 无法持续学习改进执行策略

数据要求： - 历史选品调研记录（至少 200 条完整轨迹） - 成功/失败标注（最终选品决策是否正确） - GPU 训练资源（至少 8×A100 或 4×H100）

4. 输入数据要求

请查看原始代码模板获取输入规格。

5. 输出结果

请查看原始代码模板获取输出规格。

6. 业务价值 / ROI

数据要求：高，需要大量高质量超长轨迹数据（200+ 条标注轨迹）
技术门槛：高，需要 GPU 训练基础设施 + RL 训练经验
工程复杂度：高，轨迹采集 + 分割 + SFT + RL 的完整管线
维护成本：中，模型需定期更新（随业务 SOP 变化）
长期战略价值高：自有专精 Agent 比持续依赖 API 更具竞争护城河
短期门槛较高：需要 GPU 训练资源，中小团队直接落地困难

7. 代码模板

代码块数量：6 · 路径：未检测到

"""
KLong — 超长时域 Agent 训练：轨迹分割 SFT + 渐进 RL
论文: KLong: Training LLM Agents for Extreme Long-Horizon Tasks
arXiv: 2602.17547 | 2026-02 (v2 2026-04)

核心组件:
- AgentTrajectory: 完整任务轨迹数据结构
- TrajectorySplitter: 带重叠窗口的轨迹分割器
- ProgressiveRLScheduler: 渐进式 RL 课程调度器
- ResearchFactory: 训练数据自动化生成管线
- KLongTrainer: 模拟训练流程（SFT + Progressive RL）
"""

from __future__ import annotations

from dataclasses import dataclass, field
from enum import Enum
from typing import Any


# 数据结构
@dataclass
class Step:
    """单步 Agent 交互"""
    step_id: int
    role: str                     # "user", "assistant", "tool"
    content: str
    tool_calls: list[dict] = field(default_factory=list)
    token_count: int = 0


@dataclass
class AgentTrajectory:
    """完整任务轨迹"""
    trajectory_id: str
    task_type: str
    steps: list[Step]
    total_tokens: int
    context_limit: int
    success: bool = False
    reward: float = 0.0

    @property
    def assistant_turns(self) -> int:
        return sum(1 for s in self.steps if s.role == "assistant")

    def exceeds_context(self) -> bool:
        return self.total_tokens > self.context_limit


@dataclass
class SubTrajectory:
    """分割后的子轨迹"""
    parent_id: str
    sub_id: int
    steps: list[Step]
    overlap_with_prev: int         # 与上一个子轨迹重叠的 step 数
    is_first: bool
    is_last: bool

8. 论文来源

2602.17547