AIAgent 开发者成长路线图：从使用者到架构师

传统开发者          AI Agent 时代开发者
──────────────────────────────────────────
写代码的人    →      设计 Agent 工作流的人
调 Bug 的人   →      定义质量约束和验证规则的人
做 Code Review →    训练审查 Agent 的人
手动部署      →      编排 CI/CD Agent 流水线的人
读文档        →      编写 Agent 可理解的规范的人

基础技能:
  - 理解 Prompt Engineering 基本原理
  - 能在 Claude Code/Cursor/OpenCode 中高效对话
  - 掌握基本的 Agent 交互模式（对话式、命令式、批量式）
  - 能识别 Agent 输出的正确性和局限性

工具掌握:
  - 至少精通 1 个 AI 编码工具的 CLI 或 GUI
  - 了解基本的 slash 命令和配置方式
  - 能阅读和理解 Agent 生成的代码

工作流:
  - 将 Agent 融入日常开发流程
  - 学会拆分任务让 Agent 逐步完成
  - 养成审查 Agent 输出结果的习惯

# Level 1 练习 1：用 Agent 完成一个完整功能
# 要求：不写一行代码，完全通过自然语言描述完成

# 第一步：描述需求
$ opencode "Create a Python Flask REST API for a todo list with:
             - CRUD endpoints
             - SQLite database
             - Input validation
             - Error handling
             - Unit tests"

# 第二步：审查结果
# - 检查代码结构是否合理
# - 运行测试确认功能正常
# - 阅读生成的代码，理解实现逻辑

# 第三步：迭代优化
$ opencode "Refactor the API to use async/await and add rate limiting.
             Also add OpenAPI/Swagger documentation."

# Level 1 练习 2：学会写好的 Prompt
# 差的 Prompt:
$ claude "Fix the bug in my code"

# 好的 Prompt:
$ claude "The function get_user_orders() in src/services/order_service.py
          line 45 throws a TypeError when orders is None.
          The error is: 'NoneType' object is not iterable.
          Please:
          1. Add null check at the beginning
          2. Return empty list instead of None
          3. Update the corresponding test case
          4. Explain the fix"

进阶技能:
  - 掌握 Agent 配置文件编写（CLAUDE.md, AGENTS.md 等）
  - 能编写自定义 Skills 和 Toolsets
  - 理解 MCP 协议，能配置和使用 MCP Server
  - 能为团队制定 Agent 使用规范和最佳实践

工具链:
  - 配置多 Provider 策略（按任务路由到不同模型）
  - 设置 Agent 安全边界和权限控制
  - 编写 Agent 可复用的 Prompt 模板库

自动化:
  - 将 Agent 集成到 CI/CD 流水线
  - 配置 Cron 定时任务
  - 搭建 Agent 辅助的代码审查流程

# Level 2 练习 1：编写 CLAUDE.md 项目规范
# 放在项目根目录，指导 Agent 理解项目约定

# CLAUDE.md
"""
# 项目规范

## 技术栈
- Python 3.11+, FastAPI, SQLAlchemy 2.0
- PostgreSQL 15+, Redis 7+
- pytest + httpx for testing

## 代码风格
- 使用 async/await，禁止同步阻塞调用
- 所有 public 函数必须有类型注解和 docstring
- 错误处理使用自定义异常，禁止裸 except
- 数据库操作必须通过 Repository 模式

## 测试要求
- 单元测试覆盖率 >= 80%
- 所有 API 端点必须有集成测试
- 使用 factory_boy 创建测试数据

## Git 规范
- 分支命名: feature/xxx, bugfix/xxx, hotfix/xxx
- Commit message: <type>(<scope>): <description>
- PR 必须通过 CI 检查和至少 1 人 review
"""

# Level 2 练习 2：编写自定义 MCP Server
# 让 Agent 能访问你的内部系统

from mcp.server import Server
from mcp.types import Tool, TextContent
import httpx

app = Server("internal-api-mcp")

@app.tool("query_user", description="查询内部系统的用户信息")
async def query_user(user_id: str) -> list[TextContent]:
    """通过内部 API 查询用户详情"""
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            f"https://internal-api.example.com/users/{user_id}",
            headers={"Authorization": f"Bearer {API_TOKEN}"}
        )
        return [TextContent(text=resp.text)]

@app.tool("create_ticket", description="在内部工单系统创建工单")
async def create_ticket(title: str, description: str, priority: str) -> list[TextContent]:
    """创建工单并返回工单号"""
    async with httpx.AsyncClient() as client:
        resp = await client.post(
            "https://internal-api.example.com/tickets",
            json={"title": title, "description": description, "priority": priority},
            headers={"Authorization": f"Bearer {API_TOKEN}"}
        )
        return [TextContent(text=f"工单已创建: {resp.json()['ticket_id']}")]

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

# Level 2 练习 3：配置多 Provider 路由策略

# hermes-config.yaml
providers:
  default: claude-sonnet
  routes:
    - pattern: "简单*"
      provider: claude-haiku
      reason: "低成本处理简单任务"
    - pattern: "重构*"
      provider: claude-opus
      reason: "复杂推理需要最强模型"
    - pattern: "测试*"
      provider: gpt-4o
      reason: "测试生成效果好"
    - pattern: "文档*"
      provider: claude-sonnet
      reason: "文档生成性价比最优"

budget:
  daily_limit: 50  # 每日 token 预算（百万）
  alert_threshold: 0.8  # 80% 时告警

编排技能:
  - 掌握多 Agent 架构模式（Hub-Spoke、Pipeline、P2P）
  - 能设计和实现 Agent 工作流引擎
  - 理解任务分解和依赖管理
  - 能处理 Agent 间的状态传递和错误恢复

系统设计:
  - 设计 Agent 角色和职责边界
  - 构建 Agent 通信机制
  - 实现工作流的状态机管理
  - 设计监控和可观测性方案

工程化:
  - 编写 Agent 系统的自动化测试
  - 实现工作流的版本控制
  - 搭建 Agent 系统的 CI/CD
  - 设计 Agent 行为的 A/B 测试框架

# Level 3 练习 1：实现多 Agent Code Review 流水线

import asyncio
from dataclasses import dataclass
from enum import Enum
from typing import Optional

class ReviewStatus(Enum):
    PENDING = "pending"
    STATIC_ANALYSIS = "static_analysis"
    STYLE_CHECK = "style_check"
    LOGIC_REVIEW = "logic_review"
    SECURITY_AUDIT = "security_audit"
    COMPLETED = "completed"
    FAILED = "failed"

@dataclass
class ReviewResult:
    status: ReviewStatus
    findings: list[str]
    score: float  # 0-100
    agent: str
    next_action: Optional[str] = None

class AgentPipeline:
    """多 Agent 代码审查流水线"""

    def __init__(self):
        self.agents = {
            "static": StaticAnalyzerAgent(),
            "style": StyleCheckerAgent(),
            "logic": LogicReviewerAgent(),
            "security": SecurityAuditorAgent(),
        }
        self.results: list[ReviewResult] = []

    async def execute(self, pr: PullRequest) -> ReviewResult:
        """执行完整的代码审查流水线"""

        # Stage 1: 静态分析（快速失败）
        result = await self.agents["static"].analyze(pr)
        self.results.append(result)
        if result.score < 60:
            return ReviewResult(ReviewStatus.FAILED,
                ["静态分析未通过，请先修复基础问题"], 0, "static")

        # Stage 2 & 3: 风格检查 + 逻辑审查（可并行）
        style_task = self.agents["style"].check(pr)
        logic_task = self.agents["logic"].review(pr)
        style_result, logic_result = await asyncio.gather(
            style_task, logic_task, return_exceptions=True
        )
        self.results.extend([style_result, logic_result])

        # Stage 4: 安全审计（只在前面都通过时执行）
        if style_result.score >= 70 and logic_result.score >= 70:
            security_result = await self.agents["security"].audit(pr)
            self.results.append(security_result)

        # 聚合结果
        return self._aggregate()

    def _aggregate(self) -> ReviewResult:
        """聚合所有 Agent 的审查结果"""
        all_findings = []
        for r in self.results:
            all_findings.extend(r.findings)

        avg_score = sum(r.score for r in self.results) / len(self.results)

        status = (ReviewStatus.COMPLETED if avg_score >= 70
                  else ReviewStatus.FAILED)

        return ReviewResult(status, all_findings, avg_score, "pipeline")

# Level 3 练习 2：实现 Agent 状态机和错误恢复

from enum import Enum, auto
import asyncio
import logging

class TaskState(Enum):
    CREATED = auto()
    PLANNING = auto()
    EXECUTING = auto()
    VALIDATING = auto()
    COMPLETED = auto()
    FAILED = auto()
    RETRYING = auto()

class AgentTaskRunner:
    """带状态管理和错误恢复的 Agent 任务执行器"""

    MAX_RETRIES = 3
    RETRY_DELAY = 5  # 秒

    def __init__(self, agent, validator=None):
        self.agent = agent
        self.validator = validator
        self.state = TaskState.CREATED
        self.retry_count = 0
        self.error_log = []

    async def execute(self, task: str) -> dict:
        """执行任务，包含完整的状态流转"""
        try:
            # State: PLANNING
            self.state = TaskState.PLANNING
            plan = await self.agent.plan(task)
            logging.info(f"计划: {plan}")

            # State: EXECUTING
            self.state = TaskState.EXECUTING
            result = await self.agent.execute(plan)

            # State: VALIDATING
            self.state = TaskState.VALIDATING
            if self.validator:
                validation = await self.validator.validate(result)
                if not validation.passed:
                    raise ValueError(f"验证失败: {validation.errors}")

            # State: COMPLETED
            self.state = TaskState.COMPLETED
            return result

        except Exception as e:
            self.error_log.append(str(e))
            self.state = TaskState.FAILED

            # 错误恢复
            if self.retry_count < self.MAX_RETRIES:
                self.retry_count += 1
                self.state = TaskState.RETRYING
                logging.warning(f"任务失败，{self.RETRY_DELAY}s 后重试 "
                              f"({self.retry_count}/{self.MAX_RETRIES})")
                await asyncio.sleep(self.RETRY_DELAY)
                return await self.execute(task)

            raise RuntimeError(
                f"任务最终失败，已重试 {self.MAX_RETRIES} 次。"
                f"错误日志: {self.error_log}"
            )

架构技能:
  - 设计企业级 Agent 平台的整体架构
  - 制定 Agent 工具选型和技术栈决策
  - 设计 Agent 系统的可扩展性和高可用方案
  - 构建 Agent 系统的可观测性和治理体系

战略能力:
  - 评估新技术/工具并制定采纳策略
  - 设计 Agent 系统的成本优化方案
  - 制定 AI Agent 使用的安全和合规策略
  - 规划团队 Agent 能力建设路线

技术深度:
  - 理解 LLM 的底层原理和局限性
  - 能优化 Agent 系统的性能和成本
  - 设计自定义 Agent 框架和协议
  - 能贡献开源项目或创建内部工具

# Level 4 练习：设计企业级 Agent 平台架构

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Protocol

# === 核心抽象层 ===

class AgentProvider(Protocol):
    """Agent Provider 抽象接口"""
    async def execute(self, task: str, context: dict) -> AgentResult: ...
    async def stream(self, task: str, context: dict) -> AsyncIterator[str]: ...
    @property
    def cost_per_token(self) -> float: ...
    @property
    def max_context(self) -> int: ...

class AgentOrchestrator(ABC):
    """Agent 编排器抽象基类"""

    @abstractmethod
    async def plan(self, goal: str, constraints: dict) -> Plan: ...

    @abstractmethod
    async def execute_plan(self, plan: Plan) -> ExecutionResult: ...

    @abstractmethod
    async def recover(self, failure: ExecutionFailure) -> RecoveryPlan: ...

# === 平台基础设施 ===

@dataclass
class GovernanceConfig:
    """治理配置"""
    max_tokens_per_task: int = 100_000
    max_cost_per_day: float = 100.0
    require_human_approval_for: list[str] = None
    blocked_operations: list[str] = None
    audit_log_enabled: bool = True
    data_retention_days: int = 90

class EnterpriseAgentPlatform:
    """企业级 Agent 平台"""

    def __init__(self, config: PlatformConfig):
        self.provider_registry = ProviderRegistry()
        self.orchestrator = config.orchestrator
        self.governance = GovernanceEngine(config.governance)
        self.observability = ObservabilityEngine(config.monitoring)
        self.cache = SemanticCache(config.cache)

    async def process_request(self, request: AgentRequest) -> AgentResponse:
        """处理 Agent 请求的完整流程"""

        # 1. 安全检查
        if not await self.governance.check(request):
            raise GovernanceViolation("请求未通过安全检查")

        # 2. 语义缓存查询
        cached = await self.cache.lookup(request)
        if cached:
            self.observability.record_cache_hit()
            return cached

        # 3. Provider 选择（成本 + 能力最优）
        provider = self._select_provider(request)

        # 4. 执行
        result = await provider.execute(request.task, request.context)

        # 5. 结果验证
        if not await self.governance.validate_result(result):
            raise ResultValidationError("结果未通过验证")

        # 6. 缓存和审计
        await self.cache.store(request, result)
        await self.governance.audit(request, result)
        self.observability.record_execution(request, result, provider)

        return result

    def _select_provider(self, request: AgentRequest) -> AgentProvider:
        """基于成本、能力、可用性的 Provider 选择"""
        candidates = self.provider_registry.list_available()

        # 过滤：必须满足上下文窗口要求
        candidates = [p for p in candidates
                     if p.max_context >= request.estimated_tokens]

        # 排序：综合成本和质量
        candidates.sort(key=lambda p: (
            p.cost_per_token * request.complexity_weight,
            -p.quality_score
        ))

        return candidates[0]

# Level 4 架构决策文档示例
# ADR-001: Agent 平台架构选型

## 背景
团队需要从单一的 Agent 工具使用，升级为企业级 Agent 平台，
支持多团队、多场景、可控成本的使用模式。

## 决策
采用 "编排层 + Provider 抽象层" 的架构模式：

编排层: Hermes Agent
  - 理由: 开源、活跃、编排能力强、社区支持好
  - 风险: 仍在快速发展，API 可能有 breaking changes

Provider 抽象层: 自适配层
  - 理由: 解耦业务逻辑和具体 Provider
  - 实现: 定义统一接口，每个 Provider 实现适配器

治理层: 自研
  - 理由: 企业内部的安全和合规要求无法通过开源工具满足
  - 实现: 基于策略引擎 (OPA) 的治理中间件

## 成本预估
- 开发成本: 2 人 × 2 月
- 月度运行成本: API 费用 + 服务器 ≈ $200-500
- 预期 ROI: 团队效率提升 30-50%，6 个月内收回开发成本

影响力:
  - 在 AI Agent 领域有独特见解和方法论
  - 能预判技术趋势并提前布局
  - 通过开源项目、文章、演讲影响行业

创新:
  - 发现 Agent 能力的新边界
  - 设计前所未有的 Agent 协作模式
  - 将 Agent 技术应用到新的领域

领导力:
  - 组建和培养 AI Agent 团队
  - 推动组织级的 AI Agent 转型
  - 在技术社区建立声誉和影响力

                    ┌─────────────────────────────┐
                    │      AI Agent 开发者         │
                    │        技能树全景            │
                    └──────────────┬──────────────┘
                                   │
          ┌────────────┬───────────┼───────────┬────────────┐
          ▼            ▼           ▼           ▼            ▼
    ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐
    │ LLM 基础  │ │ 工具使用  │ │ 编程基础  │ │ 系统设计  │ │ 软技能   │
    └────┬─────┘ └────┬─────┘ └────┬─────┘ └────┬─────┘ └────┬─────┘
         │             │            │             │             │
    ┌────┴─────┐  ┌────┴─────┐ ┌────┴─────┐  ┌────┴─────┐ ┌────┴─────┐
    │Prompt工程 │  │Claude Code│ │Python    │  │微服务    │ │技术写作  │
    │Token原理  │  │Codex     │ │Async     │  │事件驱动  │ │演讲     │
    │RAG基础    │  │OpenCode  │ │TypeScript│  │消息队列  │ │社区贡献  │
    │模型选择   │  │Hermes    │ │Git       │  │状态机    │ │团队管理  │
    │成本意识   │  │OpenClaw  │ │Testing   │  │可观测性  │ │战略规划  │
    └──────────┘  └──────────┘ └──────────┘  └──────────┘ └──────────┘
         │             │            │             │             │
    ┌────┴─────┐  ┌────┴─────┐ ┌────┴─────┐  ┌────┴─────┐ ┌────┴─────┐
    │Fine-tuning│ │MCP协议   │ │Docker    │  │架构模式  │ │技术布道  │
    │Embedding  │ │Skill开发  │ │K8s基础   │  │治理体系  │ │开源维护  │
    │Eval框架   │ │Workflow  │ │CI/CD     │  │成本控制  │ │标准制定  │
    └──────────┘  └──────────┘ └──────────┘  └──────────┘ └──────────┘

第 1-2 周: 基础入门
  - 注册 Claude Code / Cursor / OpenCode
  - 完成 3-5 个日常开发任务（完全用 Agent 完成）
  - 阅读 Prompt Engineering 指南
  - 写一份 CLAUDE.md / AGENTS.md 项目规范

第 3-4 周: 工具深度
  - 学习 MCP 协议，配置 2 个 MCP Server
  - 编写 3 个自定义 Skills
  - 配置多 Provider 策略
  - 将 Agent 集成到 CI/CD

第 5-6 周: 编排实践
  - 学习 Hermes Agent 的多 Agent 编排
  - 实现一个 3-Agent 协作工作流（如 Code Review 流水线）
  - 添加错误恢复和重试机制
  - 实现基本的可观测性

第 7-8 周: 系统设计
  - 研究多 Agent 架构模式
  - 设计一个企业级 Agent 平台方案
  - 编写架构决策文档 (ADR)
  - 实现治理和成本控制

第 9-10 周: 综合实战
  - 选择真实项目，端到端应用 Agent 工具链
  - 量化效率提升（速度、质量、成本）
  - 编写实践总结文档

第 11-12 周: 输出与分享
  - 在团队内部分享实践经验
  - 写一篇技术博客文章
  - 参与开源项目或社区讨论
  - 制定个人下一阶段学习目标

官方文档:
  - Claude Code 文档: docs.anthropic.com/claude-code
  - OpenCode 文档: github.com/opencode-ai/opencode
  - Hermes Agent 文档: hermes-agent.nousresearch.com
  - MCP 文档: modelcontextprotocol.io

书籍:
  - 《Designing Data-Intensive Applications》— 系统设计基础
  - 《Building Agents with LLMs》— Agent 开发指南
  - 《Prompt Engineering for LLMs》— Prompt 设计

在线课程:
  - DeepLearning.AI 的 LLM 相关课程
  - 各工具的官方 Tutorial

开源项目:
  - Hermes Agent — 学习多 Agent 编排
  - OpenClaw — 学习企业级 Agent 管理
  - MCP Server 示例 — 学习工具集成

社区:
  - GitHub Discussions (各工具)
  - Discord 社区
  - 技术博客和 Twitter/X

方向一: AI Agent 工程师
  适合: 喜欢动手实践的开发者
  职责: 设计、开发和维护 Agent 系统
  技能重点: 编程、工具链、编排
  市场需求: ⭐⭐⭐⭐⭐ 急剧增长

方向二: AI 架构师
  适合: 有系统设计经验的开发者
  职责: 设计企业级 AI 平台和技术战略
  技能重点: 架构、治理、成本控制
  市场需求: ⭐⭐⭐⭐ 稳定增长

方向三: AI 工具链开发者
  适合: 喜欢造轮子的开发者
  职责: 开发 Agent 框架、MCP Server、编排引擎
  技能重点: 底层原理、框架设计、开源贡献
  市场需求: ⭐⭐⭐⭐ 专业化需求

方向四: AI 技术顾问
  适合: 沟通能力强、有行业经验的开发者
  职责: 帮助企业制定 AI Agent 战略和落地方案
  技能重点: 行业知识、咨询能力、解决方案设计
  市场需求: ⭐⭐⭐⭐⭐ 咨询需求巨大

方向五: AI 产品经理
  适合: 产品思维强的技术人
  职责: 定义 AI Agent 产品方向和用户体验
  技能重点: 产品思维、用户研究、技术理解
  市场需求: ⭐⭐⭐⭐ 新兴岗位

2025 年 AI Agent 相关岗位薪资参考（中国一线城市）:
  AI Agent 工程师（初级）: 25K-40K/月
  AI Agent 工程师（高级）: 40K-70K/月
  AI 架构师: 60K-100K+/月
  AI 技术顾问: 50K-80K/月 + 项目奖金
  AI 产品经理: 30K-60K/月

2026 年预测:
  整体涨幅: 15-30%
  最紧缺: 多 Agent 编排工程师、AI 治理专家
  新兴岗位: Agent 可观测性工程师、AI 成本优化专家

技术关键词:
  - AI Agent 编排 / Multi-Agent Orchestration
  - MCP (Model Context Protocol)
  - Prompt Engineering / 提示词工程
  - LLM Provider 管理 / 多模型路由
  - Agent 工作流设计 / Agent Workflow Design
  - AI 治理 / AI Governance
  - 成本控制 / Cost Optimization

项目经验关键词:
  - "将 AI Agent 集成到 CI/CD 流水线，代码审查效率提升 X%"
  - "设计多 Agent 协作系统，处理 Y 种复杂场景"
  - "实现 Agent 治理平台，确保 Z 个团队安全合规使用"
  - "通过智能模型路由，月度 AI 成本降低 W%"

软实力关键词:
  - 技术选型能力
  - 架构设计能力
  - 跨团队协作
  - 技术布道

# 现在就做：
# 1. 选一个工具（推荐 OpenCode，免费开源）
# 2. 用它完成你今天的一个开发任务
# 3. 记录下：省了多少时间？发现了什么问题？

# 这就是 Level 1 的第一步。
# 每个架构师，都是从第一个 Prompt 开始的。