本报告以 2026 年奇点智能技术大会及相关公开资料为主线，围绕 Agent / AI 智能体 这条核心技术主线，系统梳理 RSI 递归自我改进、Harness 工程框架、AI4AI 元智能体、模型-数据飞轮、多智能体组织级协作、Agent 执行环境、Copilot SDK、MCP 协议、Managed Agents、RocketMQ 异步通信、具身 Robot Agent、Embedding Copilot 等方向。

报告不只做概念综述，而是以 Python 工程代码 + 架构解析 + 可落地亮点 的方式，把“智能体如何自进化、如何进入生产、如何与机器人/业务系统协同”讲清楚。

全书结构

章节内容工程产出
第1章技术主线：从 Copilot 到 Agent，再到 RSI 与 Harness架构总图
第2章RSI 递归自我改进：方法、边界与 Python 闭环RSI 实验引擎
第3章Harness 工程：Prompt/Skill/Tool/Rail 自动优化Harness Optimizer
第4章AI4AI 与模型-数据飞轮：WorkBuddy、PenguinHarness数据飞轮管道
第5章多智能体与组织级 Agent：Supervisor/Sub-Agent 架构Multi-Agent Loop
第6章Agent 执行环境：AgentENV、沙箱、长程任务、Kimi Hosted AgentsRuntime Sandbox
第7章Copilot SDK / MCP / Managed AgentsAgent Tool Gateway
第8章RocketMQ 异步底座：LiteTopic、Suspend、Consumer Group LabelAsync Agent Bus
第9章具身 Robot Agent：Embedding、Policy RSI、Harness RSIRobot Agent Loop
第10章生产级案例：店铺经营、机票履约、运维 Agent业务集成样例
第11章安全、评测、成本与治理Guardrails 模块
第12章完整工程仓库结构与后续演进可交付代码骨架

第1章 技术主线：Agent 正在从“调用模型”变成“操作系统”

1.1 为什么 Agent 成为 2026 年最核心主线

过去两年，大模型竞争主要集中在参数规模、训练数据、推理能力和上下文长度。但进入 2026 年后，行业焦点明显转向：模型如何稳定进入真实业务循环。

Agent 与普通聊天模型不同，它不是只回答问题，而是能够：

  • 感知环境；
  • 拆解任务；
  • 调用工具；
  • 读写文件；
  • 访问业务系统；
  • 与人类或其他 Agent 协作；
  • 在长程任务中持续执行；
  • 从失败轨迹中沉淀经验。

这意味着 Agent 正在从“应用层功能”升级为“系统层基础设施”。

1.2 从 Copilot 到 Agent Runtime

GitHub Copilot SDK 的出现具有标志性意义。它把 Copilot CLI 中已经经过大规模验证的 Agent 运行时开放出来，让开发者可以嵌入规划、工具调用、会话管理、流式输出等能力。

它的价值不是简单“调用 LLM”，而是提供了一套已经验证过的 Agent Loop：

User / System Event
        ↓
   Agent Runtime
        ↓
  Planning / Tool Routing
        ↓
  Tool Execution / Environment
        ↓
  Observation / State Update
        ↓
  Next Action or Final Answer

这背后体现的是 Harness 思想：模型提供智能，但真正决定 Agent 能力的，是它如何管理上下文、调用工具、验证结果、记录轨迹、恢复失败、持续迭代。

1.3 RSI：从“模型自改权重”到“系统可验证进化”

RSI，Recursive Self-Improvement，递归自我改进，传统上带有较强 AGI 色彩，指智能系统能够改进产生自身智能的机制。

但在当前工程实践中，RSI 更现实的路径不是让模型直接改写自身权重，而是让 Agent 系统持续改进：

  • Prompt；
  • Skill；
  • Tool；
  • Rail；
  • 工具调用策略；
  • 评测规则；
  • 执行流程；
  • 数据生成方式；
  • 评估器；
  • 经验库；
  • Harness 配置。

openJiuwen 提出的 RSI 框架将优化对象分为 Harness 和 Artifacts 两个维度。 Harness 决定 Agent 如何工作，Artifacts 是 Agent 产出的结果，例如论文、程序、配置或业务决策。

PenguinHarness 则进一步强调：自进化闭环必须可度量、可验证、可控制成本。

1.4 Harness 工程：RSI 的第一落点

Harness 可以理解为模型外部的运行系统。 它决定：

  • Agent 看到什么上下文；
  • 能调用哪些工具；
  • 如何拆解任务；
  • 如何验证结果；
  • 如何从失败中复盘；
  • 如何保存经验；
  • 如何进入下一轮改进。

前 OpenAI 安全副总裁 Lilian Weng 提出，自进化很可能先从 Harness 开始。 因为直接修改模型权重成本高、风险大，而 Harness 修改更可验证、更可控、更适合进入工程闭环。

但这也有边界：

  • 评估器不能太弱；
  • 上下文和记忆需要生命周期管理；
  • 负面结果容易被忽视；
  • 多样性可能坍缩；
  • Reward hacking 风险存在；
  • 长期可维护性难以自动保证。

1.5 多智能体：从个人提效到组织生产力

单 Agent 适合完成明确、封闭、可验证的任务。但真实业务往往需要：

  • 多个专业角色协作；
  • 长时间异步执行；
  • 与存量系统对接；
  • 在权限、成本、质量边界内运行；
  • 支持人工接管；
  • 支持审计与回滚。

因此，多智能体正在从“会协作”走向“可治理的组织级生产力”。

例如电商场景中，可能出现营销 Agent、客服 Agent、供应链 Agent、运营诊断 Agent 协同工作；在机票履约场景中，AI 识别、确定性工作流和 Agent 需要共同进入现有作业流程。

1.6 基础设施：AgentENV、MCP、Managed Agents、RocketMQ

越强的 Agent，越需要更强的底层系统。

  • AgentENV：面向超大规模智能体执行环境，解决长程任务、沙箱、状态管理、工具调用等问题。
  • Kimi Hosted Agents：把 Harness 作为云端服务提供，支持长程任务、多智能体协作、沙箱隔离、权限控制和审计。
  • MCP 协议：为 Agent 提供统一访问外部数据与工具的接口标准。
  • Managed Agents：提出 Brain-Hands 解耦、持久化 Session Log、事件驱动通信等架构思路。
  • RocketMQ：正在从传统消息中间件重塑为 Agent 异步通信底座，支持 LiteTopic、Consumer Group Label、Suspend 消费、MCP Server 等能力。

这些基础设施共同回答一个问题：
当 Agent 以机器速度并发执行、动态创建资源、跨系统协作时，传统面向人类节奏设计的软件架构已经不够用了。

第2章 RSI 递归自我改进：一个可运行的 Python 闭环

2.1 设计目标

本节实现一个简化版 RSI 闭环引擎，用于演示：

  • 如何收集执行轨迹；
  • 如何挖掘弱点；
  • 如何生成 Harness 改进提案；
  • 如何验证提案；
  • 如何采纳有效版本；
  • 如何沉淀经验；
  • 如何避免无限发散。

这不是生产级 RSI 系统，而是一个可用于学习、扩展和接入真实评测器的工程骨架。

2.2 核心抽象

from __future__ import annotations

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


class OptimizationTarget(str, Enum):
    HARNESS = "harness"
    ARTIFACT = "artifact"
    HYBRID = "hybrid"


@dataclass
class TaskSpec:
    """任务描述：决定 RSI 优化什么、如何评测、预算是多少。"""

    task_id: str = field(default_factory=lambda: uuid.uuid4().hex)
    name: str = ""
    target: OptimizationTarget = OptimizationTarget.HARNESS
    instruction: str = ""
    eval_fn_name: str = ""
    max_iterations: int = 5
    budget_tokens: int = 0
    constraints: Dict[str, Any] = field(default_factory=dict)


@dataclass
class HarnessConfig:
    """Harness 配置：Prompt、Skill、Tool、Rail 等。"""

    version: str = "v0"
    system_prompt: str = ""
    skills: List[str] = field(default_factory=list)
    tools: List[str] = field(default_factory=list)
    rails: List[str] = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)

    def fingerprint(self) -> str:
        raw = json.dumps(
            {
                "system_prompt": self.system_prompt,
                "skills": sorted(self.skills),
                "tools": sorted(self.tools),
                "rails": sorted(self.rails),
            },
            sort_keys=True,
        )
        return hashlib.sha256(raw.encode()).hexdigest()[:12]


@dataclass
class ExecutionResult:
    """一次任务执行结果。"""

    task_id: str
    harness_version: str
    success: bool
    score: float
    artifact: Any = None
    error: Optional[str] = None
    trace: List[Dict[str, Any]] = field(default_factory=list)
    cost_tokens: int = 0
    duration_ms: float = 0.0


@dataclass
class Weakness:
    """从失败轨迹中挖掘出的弱点。"""

    pattern: str
    severity: float
    related_tools: List[str] = field(default_factory=list)
    related_rails: List[str] = field(default_factory=list)
    suggestion: str = ""


@dataclass
class HarnessProposal:
    """Harness 改进提案。"""

    proposal_id: str = field(default_factory=lambda: uuid.uuid4().hex)
    parent_version: str = ""
    changes: Dict[str, Any] = field(default_factory=dict)
    reason: str = ""
    expected_improvement: str = ""


@dataclass
class RSIMemory:
    """RSI 经验记忆库。"""

    successes: List[Dict[str, Any]] = field(default_factory=list)
    failures: List[Dict[str, Any]] = field(default_factory=list)
    adopted_proposals: List[Dict[str, Any]] = field(default_factory=list)
    rejected_proposals: List[Dict[str, Any]] = field(default_factory=list)

2.3 RSI 闭环引擎

class RecursiveSelfImprovementEngine:
    """
    简化版 RSI 引擎。

    核心流程：
    1. 执行当前 Harness 版本；
    2. 收集轨迹与评测结果；
    3. 挖掘弱点；
    4. 生成 Harness 改进提案；
    5. 验证提案；
    6. 采纳更优版本；
    7. 沉淀经验；
    8. 判断是否继续迭代。
    """

    def __init__(
        self,
        task: TaskSpec,
        initial_harness: HarnessConfig,
        executor: Callable[[TaskSpec, HarnessConfig], ExecutionResult],
        weakness_miner: Callable, List[Weakness]],
        proposal_generator: Callable[
            [HarnessConfig, List[Weakness], RSIMemory], List[HarnessProposal]
        ],
        proposal_validator: Callable[
            [TaskSpec, HarnessConfig, HarnessProposal], ExecutionResult
        ],
    ) -> None:
        self.task = task
        self.current_harness = initial_harness
        self.executor = executor
        self.weakness_miner = weakness_miner
        self.proposal_generator = proposal_generator
        self.proposal_validator = proposal_validator
        self.memory = RSIMemory()
        self.history: List[ExecutionResult] = []
        self.best_result: Optional[ExecutionResult] = None

    def run(self) -> Dict[str, Any]:
        for iteration in range(1, self.task.max_iterations + 1):
            print(f"[RSI] iteration={iteration} harness={self.current_harness.version}")

            result = self.executor(self.task, self.current_harness)
            self.history.append(result)

            if self.best_result is None or result.score > self.best_result.score:
                self.best_result = result

            print(
                f"[RSI] score={result.score:.4f} success={result.success} "
                f"cost_tokens={result.cost_tokens} duration_ms={result.duration_ms:.1f}"
            )

            if result.success and iteration >= 2:
                print("[RSI] early stop: current harness already meets success criteria")
                break

            weaknesses = self.weakness_miner(self.history)
            if not weaknesses:
                print("[RSI] no actionable weakness found")
                break

            proposals = self.proposal_generator(
                self.current_harness, weaknesses, self.memory
            )
            if not proposals:
                print("[RSI] no valid harness proposal generated")
                break

            adopted = False
            for proposal in proposals:
                validated = self.proposal_validator(
                    self.task, self.current_harness, proposal
                )
                self.history.append(validated)

                if validated.score > (self.best_result.score if self.best_result else -1.0):
                    self.current_harness = self._apply_proposal(
                        self.current_harness, proposal
                    )
                    self.best_result = validated
                    adopted = True
                    self.memory.adopted_proposals.append(
                        {
                            "proposal_id": proposal.proposal_id,
                            "score": validated.score,
                            "reason": proposal.reason,
                        }
                    )
                    print(
                        f"[RSI] proposal adopted proposal_id={proposal.proposal_id} "
                        f"new_score={validated.score:.4f}"
                    )
                    break
                else:
                    self.memory.rejected_proposals.append(
                        {
                            "proposal_id": proposal.proposal_id,
                            "score": validated.score,
                            "reason": proposal.reason,
                        }
                    )
                    print(
                        f"[RSI] proposal rejected proposal_id={proposal.proposal_id} "
                        f"score={validated.score:.4f}"
                    )

            if not adopted:
                print("[RSI] no proposal improved the best score, stopping")
                break

        return {
            "task_id": self.task.task_id,
            "final_harness_version": self.current_harness.version,
            "best_score": self.best_result.score if self.best_result else None,
            "total_iterations": len(self.history),
            "adopted_proposals": len(self.memory.adopted_proposals),
            "rejected_proposals": len(self.memory.rejected_proposals),
        }

    @staticmethod
    def _apply_proposal(harness: HarnessConfig, proposal: HarnessProposal) -> HarnessConfig:
        new_harness = HarnessConfig(
            version=f"{harness.version}.{proposal.proposal_id[:6]}",
            system_prompt=proposal.changes.get("system_prompt", harness.system_prompt),
            skills=proposal.changes.get("skills", harness.skills),
            tools=proposal.changes.get("tools", harness.tools),
            rails=proposal.changes.get("rails", harness.rails),
            metadata={**harness.metadata, **proposal.changes.get("metadata", {})},
        )
        return new_harness

2.4 示例：用模拟评测器跑通闭环

def dummy_executor(task: TaskSpec, harness: HarnessConfig) -> ExecutionResult:
    start = time.time()

    base_score = 0.55
    prompt_bonus = min(len(harness.system_prompt) / 200, 0.15)
    skill_bonus = min(len(harness.skills) * 0.04, 0.12)
    rail_penalty = min(len(harness.rails) * 0.02, 0.08)

    score = base_score + prompt_bonus + skill_bonus - rail_penalty
    score = max(0.0, min(1.0, score))

    duration_ms = (time.time() - start) * 1000

    return ExecutionResult(
        task_id=task.task_id,
        harness_version=harness.version,
        success=score >= 0.85,
        score=round(score, 4),
        cost_tokens=120 + len(harness.skills) * 30,
        duration_ms=duration_ms,
        trace=[{"step": "execute", "harness_version": harness.version}],
    )


def dummy_weakness_miner(history: List[ExecutionResult]) -> List[Weakness]:
    weaknesses: List[Weakness] = []
    if not history:
        return weaknesses

    last = history[-1]
    if last.score < 0.85:
        weaknesses.append(
            Weakness(
                pattern="low_score_with_insufficient_skills",
                severity=0.8,
                related_tools=["search_tool", "code_runner"],
                suggestion="add structured validation skill and retry policy",
            )
        )
    return weaknesses


def dummy_proposal_generator(
    harness: HarnessConfig,
    weaknesses: List[Weakness],
    memory: RSIMemory,
) -> List[HarnessProposal]:
    proposals: List[HarnessProposal] = []
    for w in weaknesses:
        new_skills = list(harness.skills)
        if "validation_skill" not in new_skills:
            new_skills.append("validation_skill")

        proposals.append(
            HarnessProposal(
                parent_version=harness.version,
                changes={
                    "skills": new_skills,
                    "system_prompt": harness.system_prompt
                    + "\n[RSI] Add structured validation before final output.",
                },
                reason=w.suggestion,
                expected_improvement="increase success rate by adding validation skill",
            )
        )
    return proposals


def dummy_proposal_validator(
    task: TaskSpec,
    harness: HarnessConfig,
    proposal: HarnessProposal,
) -> ExecutionResult:
    candidate = RecursiveSelfImprovementEngine._apply_proposal(harness, proposal)
    return dummy_executor(task, candidate)


if __name__ == "__main__":
    task = TaskSpec(
        name="harness_optimization_demo",
        target=OptimizationTarget.HARNESS,
        instruction="Optimize agent harness for code validation tasks.",
        max_iterations=4,
    )

    initial_harness = HarnessConfig(
        version="v0",
        system_prompt="You are a helpful agent.",
        skills=["basic_coding"],
        tools=["search_tool"],
        rails=["no_shell_escape"],
    )

    engine = RecursiveSelfImprovementEngine(
        task=task,
        initial_harness=initial_harness,
        executor=dummy_executor,
        weakness_miner=dummy_weakness_miner,
        proposal_generator=dummy_proposal_generator,
        proposal_validator=dummy_proposal_validator,
    )

    report = engine.run()
    print(json.dumps(report, indent=2, ensure_ascii=False))

2.5 代码解析

这段代码的关键价值不是“跑出一个高分”，而是把 RSI 的核心工程结构显式化：

  • TaskSpec：把任务目标、评测方式、预算、约束从 Prompt 中剥离出来；
  • HarnessConfig：把 Prompt、Skill、Tool、Rail 版本化；
  • ExecutionResult：统一执行结果、轨迹、成本、耗时；
  • Weakness：把失败模式结构化；
  • HarnessProposal：把改进动作限定为可验证的小范围变更；
  • RSIMemory：记录成功经验与失败经验；
  • Engine Loop：执行 → 评测 → 挖掘 → 提案 → 验证 → 采纳 → 归档。

这符合 openJiuwen RSI 框架中“版本化优化对象、可验证执行证据、可扩展算法”的思路。

活动原文地址
在这里插入图片描述

# 人工智能 # copilot # rocketmq
Logo
DAMO开发者矩阵

DAMO开发者矩阵，由阿里巴巴达摩院和中国互联网协会联合发起，致力于探讨最前沿的技术趋势与应用成果，搭建高质量的交流与分享平台，推动技术创新与产业应用链接，围绕“人工智能与新型计算”构建开放共享的开发者生态。

更多推荐

cover

超级具身智能集群协作研究（21）：群体智能涌现与宏观秩序调控机制详解

avatar DAMO开发者矩阵
cover

超级具身智能集群协作研究（18）：基于FRA的断点续传与状态一致性恢复机制

avatar DAMO开发者矩阵
cover

超级具身智能集群协作研究（24）：基于FRA的人机回环协同与可信干预机制

avatar DAMO开发者矩阵