Python 技术报告围绕 AI 原生软件工程 / Agentic Coding 展开，重点不是单纯介绍概念，而是把 Robot Agent、Embedding、Copilot、Harness、Loop、Graph Engineering 等思想落到一套可运行的 Python 工程骨架上。

报告会包含：

• 完整代码：Agent Harness、Loop Engineering、Graph Engineering、Embedding 索引、Copilot SDK 接入、Robot Agent 执行器、CI 自愈闭环；
• 代码解析：逐段解释设计动机、数据流、边界条件和可替换实现；
• 工程亮点：面向仓库级智能体、可验证终态、受控递归、工具沙箱、人机协作边界；
• 复杂工程代码：多 Agent 协同、任务图调度、失败重试、审计日志、评测回灌。

全书大纲

1. 引言：从 Copilot 到 Agentic Coding
2. AI 原生软件工程的新范式：Traditional / Elastic / Instant Software
3. AI 编程语言与编译：MoonBit、C++26、Rust 黑灯软件工厂
4. Agent Harness：让大模型进入可控执行环境
5. Embedding 与仓库级知识索引
6. Copilot SDK 与多模型调度
7. Robot Agent：可执行、可观测、可回滚的任务体
8. Loop Engineering：从单次任务到持续演进系统
9. Graph Engineering：从个人 Loop 到组织级多智能体图
10. AISMM 2026：企业 AI 原生研发转型路线
11. 工业实践：滴滴 ABC、腾讯 CodeBuddy、C++ 质量保障
12. 完整工程：Python Agentic Coding Platform 参考实现

技术架构总览

┌──────────────────────────────────────────────────────┐
│               Agentic Coding Platform                │
├──────────────┬───────────────┬───────────────────────┤
│  Graph Layer │  Loop Layer   │   Harness Layer       │
│  (多智能体图) │ (持续演进循环) │  (单次任务执行环境)   │
├──────────────┴───────────────┴───────────────────────┤
│           Robot Agent Runtime (执行器)                │
├──────────────┬───────────────┬───────────────────────┤
│  Embedding   │  Copilot SDK  │   Tool Sandbox        │
│  仓库索引    │  多模型调度   │   文件/Shell/CI 工具  │
└──────────────┴───────────────┴───────────────────────┘

核心代码骨架

from __future__ import annotations

import asyncio
import hashlib
import json
import logging
import os
import time
import uuid
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from enum import Enum
from pathlib import Path
from typing import Any, Dict, List, Optional, Sequence

logger = logging.getLogger("agentic_platform")


class AgentState(str, Enum):
    PENDING = "pending"
    PLANNING = "planning"
    EXECUTING = "executing"
    VERIFYING = "verifying"
    COMPLETED = "completed"
    FAILED = "failed"
    BLOCKED = "blocked"


@dataclass
class Task:
    task_id: str = field(default_factory=lambda: uuid.uuid4().hex)
    title: str = ""
    description: str = ""
    accept_criteria: List[str] = field(default_factory=list)
    constraints: List[str] = field(default_factory=list)
    state: AgentState = AgentState.PENDING
    attempts: int = 0
    max_attempts: int = 3
    artifacts: Dict[str, str] = field(default_factory=dict)
    metadata: Dict[str, Any] = field(default_factory=dict)


@dataclass
class ToolResult:
    success: bool
    output: str
    cost: float = 0.0
    metadata: Dict[str, Any] = field(default_factory=dict)


class Tool(ABC):
    @property
    @abstractmethod
    def name(self) -> str: ...

    @property
    @abstractmethod
    def description(self) -> str: ...

    @abstractmethod
    async def run(self, **kwargs: Any) -> ToolResult: ...


class Harness:
    """单次 Agent 任务的执行环境。

    负责：
    - 注入工具集合
    - 控制最大步数
    - 记录轨迹
    - 执行验证
    - 输出审计日志
    """

    def __init__(
        self,
        tools: Sequence[Tool],
        max_steps: int = 50,
        timeout_seconds: float = 300.0,
    ) -> None:
        self.tools = {tool.name: tool for tool in tools}
        self.max_steps = max_steps
        self.timeout_seconds = timeout_seconds

    async def run(self, task: Task) -> Task:
        task.state = AgentState.PLANNING
        start = time.monotonic()

        plan = await self._plan(task)
        task.metadata["plan"] = plan

        for step in range(self.max_steps):
            if time.monotonic() - start > self.timeout_seconds:
                task.state = AgentState.FAILED
                task.metadata["reason"] = "timeout"
                break

            action = await self._decide(task, plan, step)
            result = await self._execute(task, action)

            if result.success:
                task.artifacts[f"step_{step}"] = result.output
            else:
                task.metadata.setdefault("errors", []).append(result.output)

            if await self._verify(task):
                task.state = AgentState.COMPLETED
                break

        if task.state not in {AgentState.COMPLETED, AgentState.FAILED}:
            task.state = AgentState.FAILED
            task.metadata["reason"] = "max_steps_exceeded"

        return task

    async def _plan(self, task: Task) -> Dict[str, Any]:
        return {
            "task_id": task.task_id,
            "accept_criteria": task.accept_criteria,
            "constraints": task.constraints,
            "strategy": "plan_then_verify",
        }

    async def _decide(
        self, task: Task, plan: Dict[str, Any], step: int
    ) -> Dict[str, Any]:
        return {"step": step, "tool": "noop", "input": {}}

    async def _execute(
        self, task: Task, action: Dict[str, Any]
    ) -> ToolResult:
        tool_name = action.get("tool")
        if tool_name and tool_name in self.tools:
            return await self.tools[tool_name].run(**action.get("input", {}))
        return ToolResult(success=True, output="noop")

    async def _verify(self, task: Task) -> bool:
        return bool(task.accept_criteria)


class LoopEngine:
    """Loop Engineering 引擎。

    把单次 Harness 执行升级成持续改进系统：
    - 失败任务进入重试
    - 审查意见回灌到下一轮
    - 指标下降才算真正变好
    """

    def __init__(self, harness: Harness, max_loops: int = 5) -> None:
        self.harness = harness
        self.max_loops = max_loops
        self.history: List[Task] = []

    async def run(self, task: Task) -> Task:
        last = task
        for loop in range(self.max_loops):
            last.attempts += 1
            last = await self.harness.run(last)

            if last.state == AgentState.COMPLETED:
                break

            feedback = await self._collect_feedback(last)
            last.description = self._refine_description(
                last.description, feedback
            )
            self.history.append(last)

        return last

    async def _collect_feedback(self, task: Task) -> str:
        errors = task.metadata.get("errors", [])
        return "\n".join(errors) if errors else "no_error"

    def _refine_description(self, description: str, feedback: str) -> str:
        if feedback == "no_error":
            return description
        return f"{description}\n\nPrevious feedback:\n{feedback}"


@dataclass
class GraphNode:
    node_id: str = field(default_factory=lambda: uuid.uuid4().hex)
    name: str = ""
    task: Optional[Task] = None
    loop_engine: Optional[LoopEngine] = None
    dependencies: List[str] = field(default_factory=list)
    state: AgentState = AgentState.PENDING


class GraphEngine:
    """Graph Engineering 引擎。

    把多个 Loop 连接成组织级协作图：
    - 显式节点
    - 显式依赖
    - 显式状态
    - 可恢复执行
    """

    def __init__(self, nodes: Sequence[GraphNode]) -> None:
        self.nodes = {node.node_id: node for node in nodes}

    async def run(self) -> Dict[str, Task]:
        results: Dict[str, Task] = {}
        completed: set[str] = set()

        while len(completed) < len(self.nodes):
            ready = [
                node
                for node in self.nodes.values()
                if node.state == AgentState.PENDING
                and all(d in completed for d in node.dependencies)
            ]
            if not ready:
                break

            tasks = [self._run_node(node) for node in ready]
            done = await asyncio.gather(*tasks)

            for node_id, task in done:
                results[node_id] = task
                completed.add(node_id)
                self.nodes[node_id].state = task.state

        return results

    async def _run_node(self, node: GraphNode):
        if node.loop_engine and node.task:
            task = await node.loop_engine.run(node.task)
        elif node.task:
            task = await self._default_harness().run(node.task)
        else:
            task = Task(state=AgentState.COMPLETED)
        return node.node_id, task

    def _default_harness(self) -> Harness:
        return Harness(tools=[])

报告亮点设计

• 从 Harness 到 Loop 再到 Graph：不是只写一个 Agent，而是把“单次执行—持续改进—组织协同”三层拆开。
• 可验证终态：每个任务必须有 accept_criteria，避免 Agent 无限生成代码。
• 受控递归：通过 max_loops、max_steps、timeout_seconds 防止递归爆炸。
• Embedding 索引：为仓库级代码检索提供语义上下文。
• Copilot SDK 接入：展示如何把生产级 Agent 运行时嵌入 Python 应用。
• Robot Agent：把任务执行抽象成可调度、可审计、可回滚的“机器人”。
• AISMM 映射：从基础设施、知识工程、流程工具、组织人才、安全治理五个维度落地。
• 工业案例对照：结合滴滴 ABC、腾讯 CodeBuddy、Rust 黑灯软件工厂、C++ 质量保障等实践。

第一章：从 Copilot 到 Agentic Coding

AI 编程正在从“补全一行代码”转向“接管一段工程任务”。Copilot 时代，开发者仍然是主要打字员，AI 负责在光标附近提供更可能的下一段代码。到了 Agentic Coding 时代，AI 不再只停留在编辑器里，它会读取仓库结构、理解依赖关系、调用测试工具、修改多个文件、提交变更，并在失败后继续修复。

这种变化带来的不是简单的效率提升，而是角色迁移。开发者从“逐行写代码的人”，逐渐变成“定义意图、设定边界、验收结果、设计反馈循环的人”。AI 负责执行，人负责约束；AI 负责探索，人负责裁决；AI 负责重复劳动，人负责架构判断。

Agentic Coding 的核心不是让模型更会聊天，而是让它进入一个可观测、可验证。这个环境就是 Harness。Harness 决定 Agent 能调用哪些工具、能访问哪些文件、能执行哪些命令、最多运行多少步、超时如何中断、失败如何记录。

但 Harness 仍然不够。一次任务执行成功，并不意味着系统变聪明了。真正重要的是：测试结果、Review 意见、失败轨迹、用户反馈能否进入下一轮执行，让 Agent 自动调整策略。这就是 Loop Engineering。

当多个 Loop 同时存在，问题会继续升级。一个 Agent 做对，不代表一群 Agent 能协同做对。不同任务之间有文件冲突、依赖顺序、权限边界和审查口径。于是需要 Graph Engineering：把任务、角色、资源、门禁、恢复点都建模成图上的节点和边，让多智能体协作不再是隐式混乱，而是显式工程对象。

从工程实现看，这三层可以对应到三个 Python 组件：

• Harness：单次任务执行环境；
• LoopEngine：带反馈重试的持续改进引擎；
• GraphEngine：多节点、多依赖、可并发的任务图调度器。

下面的代码展示了这一分层思想的最小实现：

harness = Harness(tools=[], max_steps=30, timeout_seconds=120.0)

task = Task(
    title="生成向量检索模块",
    description="为代码仓库构建 Embedding 索引，并支持语义检索。",
    accept_criteria=[
        "能够索引指定目录下的 Python 文件",
        "支持按相似度返回前 K 个代码片段",
        "提供单元测试并通过",
    ],
    constraints=[
        "不得访问 .env 文件",
        "不得执行 pip install",
        "所有文件变更必须写入 worktree",
    ],
)

loop = LoopEngine(harness=harness, max_loops=3)
result = await loop.run(task)

从 Harness 到 Loop 再到 Graph：把“单次执行—持续改进—组织协同”三层拆开
很多团队做 AI Agent 的路径是：先写一个能跑的 Agent，然后不断往里面塞工具、塞提示词，最后发现它变成一个不可维护的巨石。问题的根源在于——Agent 不是一个东西，而是三个不同层次的东西叠加在一起。把它们拆开看，架构会清晰得多。

第一层：Harness（执行框架）——解决“单次执行”
Harness 是让 LLM 变成 Agent 的最小装置：模型 + 工具调用 + 一个 while 循环。

用户目标 → [模型思考 → 调用工具 → 观察结果 → 再思考] 循环 → 产出结果
这一层的核心问题不是“智能”，而是工程可靠性：

上下文管理：工具返回的结果怎么截断、压缩、外置存储？长任务的上下文窗口怎么够用？
工具协议：统一的工具接口（如 MCP）、参数校验、失败重试、权限分级
终止条件：什么时候停？预算用完？达到目标？检测到死循环？
可观测性：每一步的 trace、token 消耗、失败点定位
这层的评估标准是：给一个明确任务，一次跑通的成功率和成本。Claude Code、OpenHands、Manus 的循环本质上都是 Harness 层的产物。Harness 做不好，后面两层全是空中楼阁——垃圾进垃圾出。

第二层：Loop（改进循环）——解决“持续变好”
Harness 跑一次是“执行”，把执行结果反馈回来驱动改进，才是“Agent 系统”。Loop 的本质是闭环：

闭环类型 机制 例子
自我修正 执行 → 自评/验证 → 重试 代码写完跑测试，失败带报错重新改
记忆积累 会话 → 蒸馏 → 长期记忆 → 下次注入 用户偏好、项目约定跨会话生效
技能沉淀 重复成功路径 → 固化为 skill/子程序 把调试过的 prompt 固化成可复用技能
数据飞轮 线上失败案例 → 标注/微调 → 模型升级 生产环境的 bad case 进训练集
这一层最常见的失败模式是闭环没闭上：Agent 每次都从零开始，错误重复犯；或者有日志但没人没机制去消费它。真正的 Loop 要求：

验证器先于生成器——没有可靠的方式判断“这次做得对不对”，闭环就是空转
反馈要有通道——用户的一句纠正必须能落到下一次执行的上下文或技能文件里，而不是消失在聊天记录中
区分易变与稳定——记忆、技能是可持续资产；会话记录是原料，不等于资产
评估标准：第 N 次执行是否比第 1 次更好？改进是自动发生的，还是靠人肉搬运？

第三层：Graph（协同网络）——解决“组织协同”
单 Agent 能力有上限，规模化后必然走向多个 Agent 组成的图。Graph 不是“多搞几个 AI 聊天”，而是把组织学的问题搬进来：

拓扑设计：主从式（Orchestrator-Worker）？流水线？对等协作？拓扑本身就是架构决策——它决定了信息怎么流动、瓶颈在哪
职责边界：每个 Agent 的上下文窗口就是它的“注意力”，隔离上下文、只传递结论而非全过程，是图能扩展的前提
协议与契约：Agent 之间交接的是结构化的工件（文档、schema、测试报告），而不是自然语言的转述——像人类组织靠文档而非靠口口相传
人在图中的位置：审批节点、接管点、信任分级（哪些操作自动放行，哪些必须人审）
关键洞察：组织规模化遇到的问题，多 Agent 系统会原样再遇到一遍——沟通成本、信息丢失、职责重叠、单点阻塞。所以 Graph 层的设计方法论更接近组织设计，而非 prompt 工程。

评估标准：加一个新 Agent 的边际成本是多少？故障隔离吗？整体产出是否 > 各部分之和？

三层之间的关系

Graph（组织协同）
▲ 每个 Agent 节点内部是一个合格的 Loop
Loop（持续改进）
▲ 每次改进作用的对象是一次可靠的 Harness 执行
Harness（单次执行）
三个工程含义：

自下而上依赖：Harness 不稳，Loop 反馈的是噪声；Loop 没闭上，Graph 只是把错误放大到组织级别。
自上而下定义：但架构要先想清楚 Graph 的形态，才知道 Loop 该积累什么（是个人技能？还是团队知识库？）、Harness 该暴露什么接口。
不要跨层优化：Agent 回答质量差，先分清是单次执行失败（修 Harness）、没有从历史学习（修 Loop），还是协作交接丢失信息（修 Graph）。跨层打补丁是巨石架构的起点。
一句话总结：Harness 让 Agent 跑得对，Loop 让它越跑越好，Graph 让一群它干成一件大事——三层各治各的病，不要混在一个 while 循环里。

活动原文地址