原创解读：Kaggle 白皮书《Introduction to Agents》——AI Agent 入门与架构全景

📋 版权声明与免责声明

本文是作者基于 Kaggle 白皮书《Introduction to Agents》的原创分析文章。

观点归属声明：

• 文中关于 Agent 五个层级、架构演进与工程实践的分析为作者基于白皮书的独立整理与再阐释
• 行文结构、案例组织与重点判断体现作者个人理解
• 文章并非对白皮书的逐段翻译，而是面向工程实践的原创解读

原文参考：

• 标题：《Introduction to Agents》
• 作者：Alan Blount, Antonio Gulli, Shubham Saboo, Michael Zimmermann, Vladimir Vuskovic
• 链接：阅读原文

原创性质：本文为独立创作的解读类文章，非翻译或改写作品。文中观点仅代表作者个人理解，与原文作者立场可能不同。

---

引子：从会答题的模型，到会完成任务的 Agent

过去几年，大模型最让人惊艳的能力，主要体现在“回答问题”和“生成内容”上：它们能写代码、做总结、生成文案，也能在很多场景中给出近似专家级的解释。但如果把视角从“生成一个回答”切换到“完成一个目标”，我们会立刻发现传统 LLM 的边界——它能提供建议，却未必能主动规划、调用工具、分解任务并一路执行到结果。

Kaggle 白皮书《Introduction to Agents》真正有价值的地方，就在于它把这种差异讲清楚了：Agent 不是把 LLM 接进某个工作流里那么简单，而是一种能够围绕目标持续决策与行动的完整应用系统。

这篇文章不会按白皮书顺序逐段复述，而是站在工程实践的视角，把其中最值得关注的部分重新组织出来：为什么 Agent 是一次范式转移？如何理解它的五个层级？从基础推理、工具连接到多步骤规划、系统协作，架构复杂度是怎样一步步上升的？以及，这些概念为什么会直接影响你未来构建 AI 产品的方式？

---

一、从预测式 AI 到自主 Agent：范式转移

过去几年，AI 的发展重心始终围绕被动式、离散任务的模型——翻译一段话、回答一个问题、根据提示生成一张图片。这种范式虽然强大，但每一步都需要人类明确的指令。

现在我们正在见证一场范式转移：从仅能预测或创造内容的 AI，向能够自主解决问题和执行任务的新型软件演进。

这就是 AI Agent 的核心理念。

1.1 范式对比：从被动到主动

让我们通过具体场景来理解这种差异：

传统 AI（预测式）：

用户：北京明天天气怎么样？
AI：北京明天多云，气温15-22度。
（对话结束，等待下一个指令）

Agent（自主式）：

用户：我想周末去北京玩，有什么建议？
Agent：
1. 查询北京周末天气预报
2. 搜索北京周末热门景点
3. 根据天气推荐适合的户外活动
4. 查询交通信息并建议出行方式
5. 综合以上信息生成完整攻略

这是一个完整的建议：考虑到周六晴天，推荐您早上去长城，下午逛颐和园...

Agent 不只是回答问题，而是主动规划并执行一系列动作来达成目标。

1.2 Agent 究竟是什么？

很多人误以为 Agent 就是把 LLM 嵌入到工作流中。但在 Kaggle 这篇 54 页的权威白皮书中，作者们给出了更精确的定义：

Agent 不是一个静态工作流中的 AI 模型，而是一个完整的应用系统——它能够制定计划并采取行动以实现目标。

Agent 将语言模型的推理能力与实际行动能力相结合，使其能够处理模型本身无法完成的复杂、多步骤任务。

关键能力在于：Agent 可以独立工作，自主 figuring out 达成目标所需的后续步骤，无需人类在每一步都进行指导。

1.3 为什么现在 Agent 变得重要？

有三个关键因素推动了 Agent 的崛起：

1. 模型能力飞跃：GPT-4、Claude 等大模型具备了强大的推理和规划能力
2. 工具生态成熟：Function Calling、MCP 等协议让模型与外部世界连接变得标准化
3. 实际需求驱动：企业需要能够自动化复杂业务流程的 AI 系统

---

二、Agent 的五个层级：从推理到自我进化

白皮书提出了一个非常有价值的分类体系（Taxonomy），将 Agent 系统分为 5 个层级。这个分类不仅帮助我们理解 Agent 的演进路径，也为架构设计提供了清晰的参考框架。

Level 0: 核心推理系统 (Core Reasoning System)

这是基础层，只包含语言模型本身。它能推理、回答问题，但没有与外部世界交互的能力。

能力边界：

• 可以回答知识性问题
• 可以进行逻辑推理
• 可以生成文本和代码
• 无法获取实时信息
• 无法执行外部操作

典型应用：

• 内部知识问答（基于训练数据）
• 文本生成和润色
• 简单的逻辑推理任务

架构示例：

用户输入 → LLM → 模型输出

代码示例：

from openai import OpenAI

client = OpenAI()

class Level0Agent:
    def __init__(self):
        self.system_prompt = "你是一个有帮助的助手。"

    def chat(self, user_message):
        response = client.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": self.system_prompt},
                {"role": "user", "content": user_message}
            ]
        )
        return response.choices[0].message.content

Level 1: 连接型问题解决者 (Connected Problem-Solver)

在 Level 0 基础上增加了工具调用能力。Agent 可以搜索网页、查询数据库、调用 API 来获取信息或执行操作。

核心改进：

• 能够访问实时信息
• 可以调用外部 API
• 具备基础的数据获取能力

典型应用：

• 带搜索功能的问答助手
• 天气查询机器人
• 股票信息查询

架构示例：

用户输入 → LLM → 需要工具？ → 调用工具 → 整合结果 → 输出
                ↓ 不需要      ↑
                └────────→ 直接输出

代码示例：

import json
from openai import OpenAI

client = OpenAI()

class Level1Agent:
    def __init__(self):
        self.tools = [
            {
                "type": "function",
                "function": {
                    "name": "search_web",
                    "description": "搜索网页获取实时信息",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "query": {"type": "string"}
                        },
                        "required": ["query"]
                    }
                }
            },
            {
                "type": "function",
                "function": {
                    "name": "get_weather",
                    "description": "获取指定城市的天气",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "city": {"type": "string"},
                            "date": {"type": "string"}
                        },
                        "required": ["city"]
                    }
                }
            }
        ]

    def execute_tool(self, tool_name, params):
        if tool_name == "get_weather":
            # 实际实现会调用天气 API
            return {"temperature": 22, "condition": "sunny"}
        return {}

    def chat(self, user_message):
        messages = [{"role": "user", "content": user_message}]

        # 第一次调用，获取工具调用请求
        response = client.chat.completions.create(
            model="gpt-4",
            messages=messages,
            tools=self.tools,
            tool_choice="auto"
        )

        message = response.choices[0].message

        # 检查是否需要调用工具
        if message.tool_calls:
            # 执行工具调用
            tool_results = []
            for tool_call in message.tool_calls:
                function_name = tool_call.function.name
                function_params = json.loads(tool_call.function.arguments)
                result = self.execute_tool(function_name, function_params)
                tool_results.append({
                    "tool_call_id": tool_call.id,
                    "role": "tool",
                    "name": function_name,
                    "content": json.dumps(result)
                })

            # 将工具结果添加到对话
            messages.append(message)
            messages.extend(tool_results)

            # 第二次调用，获取最终回复
            final_response = client.chat.completions.create(
                model="gpt-4",
                messages=messages
            )
            return final_response.choices[0].message.content

        return message.content

Level 2: 策略型问题解决者 (Strategic Problem-Solver)

这一层引入了规划能力和多步骤推理。Agent 能够将复杂任务分解为子任务，并按策略顺序执行。

核心改进：

• 任务分解 (Task Decomposition)
• 多步骤规划
• 错误恢复机制
• 中间结果的处理和整合

典型应用：

• 研究报告生成（搜索→分析→撰写）
• 数据分析任务（获取数据→清洗→分析→可视化）
• 复杂问题诊断

架构示例：

用户目标 → 任务分解器 → 子任务1 → 执行 → 结果
                     ↓
                     ├→ 子任务2 → 执行 → 结果
                     ↓
                     ├→ 子任务3 → 执行 → 结果
                     ↓
                     结果整合器 → 最终输出

代码示例：

class Level2Agent:
    def __init__(self):
        self.level1_agent = Level1Agent()

    def decompose_task(self, goal):
        """将复杂任务分解为子任务"""
        decomposition_prompt = f"""
        请将以下目标分解为具体的执行步骤：
        目标：{goal}

        要求：
        1. 每个步骤应该是可执行的
        2. 步骤之间要有明确的依赖关系
        3. 输出格式为 JSON 数组

        示例输出：
        [
            {"step": 1, "action": "搜索相关资料", "depends_on": []},
            {"step": 2, "action": "分析收集的信息", "depends_on": [1]},
            {"step": 3, "action": "生成报告", "depends_on": [2]}
        ]
        """

        response = client.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": decomposition_prompt}]
        )

        steps = json.loads(response.choices[0].message.content)
        return steps

    def execute_task(self, goal):
        """执行复杂任务"""
        # 1. 任务分解
        steps = self.decompose_task(goal)

        # 2. 按依赖顺序执行
        results = {}
        for step in steps:
            step_num = step["step"]
            action = step["action"]

            # 构建上下文（包含之前步骤的结果）
            context = f"任务：{action}\n"
            if step.get("depends_on"):
                for dep in step["depends_on"]:
                    context += f"步骤{dep}的结果：{results.get(dep, '')}\n"

            # 执行步骤
            result = self.level1_agent.chat(context)
            results[step_num] = result

            print(f"步骤 {step_num} 完成: {action}")

        # 3. 整合结果
        final_result = self.synthesize_results(goal, results)
        return final_result

    def synthesize_results(self, goal, results):
        """整合所有步骤的结果"""
        synthesis_prompt = f"""
        目标：{goal}

        各步骤执行结果：
        {json.dumps(results, ensure_ascii=False, indent=2)}

        请整合以上结果，生成最终输出。
        """

        response = client.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": synthesis_prompt}]
        )

        return response.choices[0].message.content

Level 3: 协作式多 Agent 系统 (Collaborative Multi-Agent System)

多个 Agent 协同工作，每个 Agent 可能 specialized 于特定领域。它们可以分工合作、相互通信、协调复杂的并行任务。

核心改进：

• 专业化分工
• Agent 间通信
• 并行执行
• 结果协调与整合

类比：就像一个软件开发团队，有产品经理、后端工程师、前端工程师、测试人员各自负责不同环节。

典型应用：

• 软件开发助手（需求分析→设计→编码→测试）
• 投资研究团队（市场分析→财务分析→风险评估→投资建议）
• 内容创作团队（策划→写作→编辑→设计）

架构示例：

                    ┌→ 研究 Agent → 数据
用户请求 → 协调 Agent ┼→ 分析 Agent → 洞察
                    ├→ 写作 Agent → 内容
                    └→ 审核 Agent → 质检
                              ↓
                         整合输出

代码示例：

class SpecializedAgent:
    """专业化 Agent 基类"""
    def __init__(self, name, expertise):
        self.name = name
        self.expertise = expertise

    def process(self, task, context=None):
        prompt = f"""
        你是{self.name}，专长于{self.expertise}。

        任务：{task}
        {f"相关背景：{context}" if context else ""}

        请基于你的专业领域完成此任务。
        """

        response = client.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": prompt}]
        )

        return {
            "agent": self.name,
            "result": response.choices[0].message.content
        }

class Level3MultiAgent:
    def __init__(self):
        self.coordinator = SpecializedAgent("协调员", "任务分解与协调")
        self.researcher = SpecializedAgent("研究员", "信息收集与研究")
        self.analyst = SpecializedAgent("分析师", "数据分析与洞察")
        self.writer = SpecializedAgent("撰稿人", "内容创作与写作")

    def execute(self, goal):
        """多 Agent 协作执行"""
        # 1. 协调员分解任务并分配
        coordination_prompt = f"""
        目标：{goal}

        可用 Agent：
        - 研究员：负责信息收集
        - 分析师：负责数据分析
        - 撰稿人：负责内容创作

        请制定协作计划，明确每个 Agent 的任务和输入输出。
        """

        plan_response = client.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": coordination_prompt}]
        )
        plan = plan_response.choices[0].message.content

        # 2. 并行执行（简化示例，实际可真正的并行）
        research_result = self.researcher.process(f"研究主题：{goal}")

        # 研究员完成后，分析师开始工作
        analysis_result = self.analyst.process(
            f"分析以下研究结果：{research_result['result']}",
            context=research_result['result']
        )

        # 最后撰稿人整合
        final_result = self.writer.process(
            f"基于以下分析撰写最终报告：{analysis_result['result']}",
            context=f"研究：{research_result['result']}\n分析：{analysis_result['result']}"
        )

        return {
            "plan": plan,
            "research": research_result,
            "analysis": analysis_result,
            "final": final_result
        }

Level 4: 自我进化系统 (Self-Evolving System)

最高层级，Agent 能够从经验中学习、自我优化策略、改进自己的工具使用方式，甚至修改自身的系统配置。

核心改进：

• 从经验中学习
• 自我优化策略
• 工具使用改进
• 系统配置自适应

这代表了 Agent 技术的终极愿景——真正能够持续改进的自主系统。

典型应用：

• 自我优化的客服系统
• 自适应交易策略
• 持续学习的科研助手

核心机制：

执行 → 观察结果 → 分析性能 → 识别改进点 → 调整策略 → 再次执行
         ↑                                                  ↓
         └──────────────── 反馈循环 ←───────────────────────┘

---

三、核心架构解析：模型、工具与编排层

白皮书详细阐述了 Agent 的三大核心组件。理解这三者的关系，是设计 Agent 系统的关键。

3.1 模型：Agent 的”大脑”

模型负责 Agent 的”思考”过程：

核心职责：

• 推理：理解问题、分析情况、进行逻辑推导
• 规划：制定行动步骤、确定执行顺序
• 决策：选择使用哪些工具、何时停止
• 反思：评估执行结果、识别错误并修正

模型选型矩阵：

场景	推荐模型	理由
复杂推理任务	GPT-4, Claude-3 Opus	强大的多步推理能力
代码生成	Claude-3.5 Sonnet, GPT-4	代码理解和生成能力强
成本敏感场景	GPT-3.5, Claude-3 Haiku	性价比高
长上下文处理	Claude-3 (200K), GPT-4 Turbo (128K)	大上下文窗口
实时交互	GPT-4 Turbo	低延迟

提示词工程原则：

1. 角色定义清晰：明确 Agent 的身份和职责边界
2. 上下文充分：提供足够的背景信息和约束条件
3. 输出格式规范：使用结构化输出便于后续处理
4. 示例引导：提供输入输出示例帮助模型理解

3.2 工具：Agent 的”手”

工具使 Agent 能够与世界交互，是 Agent 能力的延伸。

信息获取类工具（Grounding）：

# 搜索引擎
search_tool = {
    "name": "web_search",
    "description": "搜索互联网获取实时信息",
    "parameters": {
        "query": "搜索关键词",
        "num_results": "返回结果数量"
    }
}

# 数据库查询
database_tool = {
    "name": "query_database",
    "description": "查询内部数据库",
    "parameters": {
        "sql": "SQL查询语句",
        "database": "目标数据库"
    }
}

# RAG 检索
rag_tool = {
    "name": "retrieve_knowledge",
    "description": "从知识库检索相关信息",
    "parameters": {
        "query": "检索查询",
        "top_k": "返回结果数"
    }
}

操作执行类工具（Action）：

# 发送邮件
email_tool = {
    "name": "send_email",
    "description": "发送邮件",
    "parameters": {
        "to": "收件人",
        "subject": "主题",
        "body": "内容"
    }
}

# 创建日历事件
calendar_tool = {
    "name": "create_event",
    "description": "创建日历事件",
    "parameters": {
        "title": "事件标题",
        "start_time": "开始时间",
        "end_time": "结束时间"
    }
}

# API 调用
generic_api_tool = {
    "name": "call_api",
    "description": "调用任意API",
    "parameters": {
        "url": "API地址",
        "method": "HTTP方法",
        "headers": "请求头",
        "body": "请求体"
    }
}

工具设计最佳实践：

1. 原子性：每个工具只做一件事，做好一件事
2. 幂等性：相同输入产生相同输出，便于重试
3. 自描述性：工具描述要清晰表达其用途和参数
4. 错误处理：定义清晰的错误类型和处理方式

3.3 编排层（Orchestration Layer）：Agent 的”神经系统”

编排层是 Agent 系统的核心，负责协调模型和工具的交互。

核心职责：

• 对话状态管理：维护多轮对话的上下文
• 工具调用编排：处理工具调用的输入/输出转换
• 上下文窗口管理：确保不超出模型的上下文限制
• 错误处理和重试：优雅地处理各种错误情况
• 人机协作交接：在需要时寻求人类确认

编排层架构：

┌─────────────────────────────────────────────────────────┐
│                    Orchestration Layer                  │
├─────────────────────────────────────────────────────────┤
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐  │
│  │   Context    │  │    Tool      │  │    Error     │  │
│  │   Manager    │  │   Executor   │  │   Handler    │  │
│  └──────────────┘  └──────────────┘  └──────────────┘  │
├─────────────────────────────────────────────────────────┤
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐  │
│  │   Session    │  │   Planning   │  │   Human      │  │
│  │    Store     │  │    Engine    │  │   Handoff    │  │
│  └──────────────┘  └──────────────┘  └──────────────┘  │
└─────────────────────────────────────────────────────────┘

代码示例 - 完整编排器：

import asyncio
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
from enum import Enum

class TaskStatus(Enum):
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    FAILED = "failed"

@dataclass
class Task:
    id: str
    description: str
    status: TaskStatus
    result: Optional[Any] = None
    error: Optional[str] = None

class Orchestrator:
    """Agent 编排器"""

    def __init__(self, llm_client, tools, session_store):
        self.llm = llm_client
        self.tools = {tool["function"]["name"]: tool for tool in tools}
        self.session_store = session_store
        self.max_iterations = 10  # 防止无限循环

    async def execute(self, user_input: str, session_id: str) -> str:
        """执行用户请求"""
        # 1. 加载会话上下文
        context = await self.load_context(session_id)

        # 2. 构建消息历史
        messages = context + [{"role": "user", "content": user_input}]

        # 3. 执行循环
        iteration = 0
        while iteration < self.max_iterations:
            iteration += 1

            # 调用 LLM
            response = await self.call_llm(messages)

            # 检查是否完成任务
            if not response.tool_calls:
                # 没有工具调用，直接返回结果
                await self.save_context(session_id, messages + [response])
                return response.content

            # 执行工具调用
            tool_results = await self.execute_tools(response.tool_calls)

            # 添加结果到消息历史
            messages.extend([response] + tool_results)

            # 检查是否需要人工介入
            if await self.needs_human_handoff(messages):
                return await self.request_human_handoff(session_id, messages)

        # 超出最大迭代次数
        return "任务执行时间过长，请简化您的请求或稍后重试。"

    async def call_llm(self, messages: List[Dict]) -> Any:
        """调用 LLM"""
        return self.llm.chat.completions.create(
            model="gpt-4",
            messages=messages,
            tools=list(self.tools.values()),
            tool_choice="auto"
        ).choices[0].message

    async def execute_tools(self, tool_calls: List[Any]) -> List[Dict]:
        """执行工具调用"""
        results = []

        for tool_call in tool_calls:
            function_name = tool_call.function.name
            function_args = json.loads(tool_call.function.arguments)

            try:
                # 实际工具执行逻辑
                result = await self.run_tool(function_name, function_args)
                results.append({
                    "tool_call_id": tool_call.id,
                    "role": "tool",
                    "name": function_name,
                    "content": json.dumps(result)
                })
            except Exception as e:
                results.append({
                    "tool_call_id": tool_call.id,
                    "role": "tool",
                    "name": function_name,
                    "content": json.dumps({"error": str(e)})
                })

        return results

    async def run_tool(self, name: str, args: Dict) -> Any:
        """运行具体工具"""
        # 根据工具名称分发到具体实现
        tool_implementations = {
            "search_web": self.search_web,
            "get_weather": self.get_weather,
            # ... 更多工具
        }

        if name in tool_implementations:
            return await tool_implementations[name](**args)
        else:
            raise ValueError(f"Unknown tool: {name}")

    async def needs_human_handoff(self, messages: List[Dict]) -> bool:
        """判断是否需要人工介入"""
        # 实现判断逻辑，例如：
        # - 检测到敏感操作
        # - 多次重试失败
        # - 用户明确要求人工
        return False

    async def load_context(self, session_id: str) -> List[Dict]:
        """加载会话上下文"""
        return await self.session_store.get(session_id, default=[])

    async def save_context(self, session_id: str, messages: List[Dict]):
        """保存会话上下文"""
        # 截断过长的上下文
        truncated = self.truncate_context(messages)
        await self.session_store.set(session_id, truncated)

    def truncate_context(self, messages: List[Dict], max_tokens: int = 8000) -> List[Dict]:
        """截断上下文以符合 token 限制"""
        # 简化实现：保留系统消息和最近的消息
        system_msgs = [m for m in messages if m.get("role") == "system"]
        other_msgs = [m for m in messages if m.get("role") != "system"]

        # 保留最近的消息
        keep_count = min(len(other_msgs), 10)  # 简化逻辑
        return system_msgs + other_msgs[-keep_count:]

---

四、设计选择的关键考量

4.1 领域知识与角色设定

为你的 Agent 设定清晰的角色（Persona）和领域知识至关重要：

角色设定模板：

你是一个专业的{角色}，专长于{领域}。

你的职责：
1. {职责1}
2. {职责2}
3. {职责3}

你的行为准则：
- {准则1}
- {准则2}
- {准则3}

你的限制：
- 你不能{限制1}
- 你不能{限制2}

示例 - 医疗咨询 Agent：

你是一个专业的医疗健康助理，具有医学背景知识。

你的职责：
1. 回答用户关于健康、疾病、用药的一般性问题
2. 提供健康生活方式的建议
3. 帮助用户理解医学术语和检查报告

你的行为准则：
- 回答要基于循证医学知识
- 使用通俗易懂的语言解释专业概念
- 保持同理心和耐心

你的限制：
- 你不能提供诊断，必须建议用户咨询专业医生
- 你不能开具处方或推荐具体药物剂量
- 你不能处理紧急医疗情况，必须建议拨打急救电话

4.2 上下文增强策略

短期记忆管理：

class ShortTermMemory:
    def __init__(self, max_messages=20):
        self.messages = []
        self.max_messages = max_messages

    def add(self, message):
        self.messages.append(message)
        if len(self.messages) > self.max_messages:
            self.messages = self.messages[-self.max_messages:]

    def get_context(self):
        return self.messages

    def clear(self):
        self.messages = []

长期记忆实现：

import hashlib

class LongTermMemory:
    def __init__(self, vector_store):
        self.vector_store = vector_store

    async def remember(self, user_id: str, key: str, value: str, importance: float = 1.0):
        """存储长期记忆"""
        memory_id = hashlib.md5(f"{user_id}:{key}".encode()).hexdigest()

        await self.vector_store.upsert(
            ids=[memory_id],
            documents=[value],
            metadatas=[{
                "user_id": user_id,
                "key": key,
                "importance": importance,
                "timestamp": time.time()
            }]
        )

    async def recall(self, user_id: str, query: str, top_k: int = 5):
        """检索相关记忆"""
        results = await self.vector_store.query(
            query_texts=[query],
            where={"user_id": user_id},
            n_results=top_k
        )
        return results

4.3 多 Agent 系统设计模式

白皮书提到了几种常见的多 Agent 架构：

1. 主从式（Master-Worker）：

Master Agent
    ├─ Worker 1 (数据处理)
    ├─ Worker 2 (分析)
    └─ Worker 3 (报告生成)

适用场景：任务可以明确分解为子任务

2. 流水线式（Pipeline）：

Input → Agent A → Agent B → Agent C → Output
         (提取)    (分析)    (生成)

适用场景：数据处理流程固定的场景

3. 竞争式（Competitive）：

Input ┬→ Agent A ─┐
      ├→ Agent B ─┼→ Selector → Best Output
      └→ Agent C ─┘

适用场景：需要多方案择优的场景

4. 协作式（Collaborative）：

         ┌←→┐
Agent A ─┤  ├─ Agent B
         └←→┘
    ↕       ↕
Agent C ←→ Agent D

适用场景：复杂问题需要多方协商

---

五、Agent 运维：结构化地应对不确定性

与传统软件不同，Agent 的行为具有非确定性。白皮书提出了”Agent Ops”的概念。

5.1 测量关键指标

核心指标体系：

指标类别	具体指标	计算方式	目标值
任务完成	成功率	成功任务/总任务	> 90%
效率	平均步骤数	总步骤/完成任务	< 5 步
成本	单任务成本	Token 费用 + API 费用	< $0.1
延迟	平均响应时间	总时间/请求数	< 3s
质量	用户满意度	评分平均值	> 4.0/5

指标监控代码示例：

from dataclasses import dataclass
from typing import List
import time

@dataclass
class AgentMetrics:
    total_requests: int = 0
    successful_requests: int = 0
    total_steps: int = 0
    total_latency: float = 0
    total_cost: float = 0
    user_ratings: List[float] = None

    def __post_init__(self):
        if self.user_ratings is None:
            self.user_ratings = []

    @property
    def success_rate(self):
        if self.total_requests == 0:
            return 0
        return self.successful_requests / self.total_requests

    @property
    def avg_steps(self):
        if self.successful_requests == 0:
            return 0
        return self.total_steps / self.successful_requests

    @property
    def avg_latency(self):
        if self.total_requests == 0:
            return 0
        return self.total_latency / self.total_requests

    @property
    def avg_cost(self):
        if self.total_requests == 0:
            return 0
        return self.total_cost / self.total_requests

    @property
    def avg_rating(self):
        if not self.user_ratings:
            return 0
        return sum(self.user_ratings) / len(self.user_ratings)

    def report(self):
        return f"""
        Agent 性能报告
        ==============
        总请求数：{self.total_requests}
        成功率：{self.success_rate:.2%}
        平均步骤数：{self.avg_steps:.1f}
        平均延迟：{self.avg_latency:.2f}s
        平均成本：${self.avg_cost:.4f}
        用户满意度：{self.avg_rating:.2f}/5
        """

5.2 A/B 测试思维

将 Agent 视为一个需要持续实验的系统：

A/B 测试框架：

import random

class ABTestFramework:
    def __init__(self):
        self.experiments = {}

    def create_experiment(self, name, variants, traffic_split):
        """
        创建实验
        variants: {'control': control_config, 'treatment': treatment_config}
        traffic_split: {'control': 0.5, 'treatment': 0.5}
        """
        self.experiments[name] = {
            'variants': variants,
            'traffic_split': traffic_split,
            'metrics': {'control': [], 'treatment': []}
        }

    def get_variant(self, experiment_name, user_id):
        """根据用户 ID 分配实验组"""
        exp = self.experiments[experiment_name]

        # 使用用户 ID 哈希确保一致性
        hash_val = hash(f"{experiment_name}:{user_id}") % 100

        cumulative = 0
        for variant, split in exp['traffic_split'].items():
            cumulative += split * 100
            if hash_val < cumulative:
                return variant, exp['variants'][variant]

        return 'control', exp['variants']['control']

    def record_metric(self, experiment_name, variant, metric_value):
        """记录实验指标"""
        self.experiments[experiment_name]['metrics'][variant].append(metric_value)

    def analyze_results(self, experiment_name):
        """分析实验结果"""
        metrics = self.experiments[experiment_name]['metrics']

        control_metrics = metrics['control']
        treatment_metrics = metrics['treatment']

        # 计算统计显著性
        from scipy import stats
        t_stat, p_value = stats.ttest_ind(control_metrics, treatment_metrics)

        control_mean = sum(control_metrics) / len(control_metrics)
        treatment_mean = sum(treatment_metrics) / len(treatment_metrics)
        lift = (treatment_mean - control_mean) / control_mean

        return {
            'control_mean': control_mean,
            'treatment_mean': treatment_mean,
            'lift': lift,
            'p_value': p_value,
            'significant': p_value < 0.05
        }

5.3 LLM 作为评判者

使用另一个 LLM 来评估 Agent 的输出质量：

class LLMJudge:
    def __init__(self, judge_model="gpt-4"):
        self.judge_model = judge_model

    async def evaluate(self, task_description, agent_output, criteria):
        """评估 Agent 输出"""
        prompt = f"""
        你是一个公正的评判者。请评估以下 Agent 输出的质量。

        任务描述：
        {task_description}

        Agent 输出：
        {agent_output}

        评估维度：
        {criteria}

        请按以下 JSON 格式输出评估结果：
        {{
            "overall_score": 1-5,
            "dimension_scores": {{
                "准确性": 1-5,
                "完整性": 1-5,
                "清晰度": 1-5,
                "有用性": 1-5
            }},
            "reasoning": "详细说明评分的理由",
            "strengths": ["优点1", "优点2"],
            "weaknesses": ["不足1", "不足2"],
            "suggestions": ["改进建议1", "改进建议2"]
        }}
        """

        response = await self.call_llm(prompt)
        return json.loads(response)

    async def batch_evaluate(self, test_cases):
        """批量评估"""
        results = []
        for case in test_cases:
            result = await self.evaluate(
                case['task'],
                case['output'],
                case['criteria']
            )
            results.append(result)
        return results

5.4 可观测性

使用 OpenTelemetry 追踪 Agent 的执行路径：

from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor

# 配置追踪
trace.set_tracer_provider(TracerProvider())
tracer = trace.get_tracer(__name__)

otlp_exporter = OTLPSpanExporter(endpoint="your-collector-endpoint")
span_processor = BatchSpanProcessor(otlp_exporter)
trace.get_tracer_provider().add_span_processor(span_processor)

class ObservableAgent:
    @tracer.start_as_current_span("agent_execution")
    async def execute(self, user_input):
        with tracer.start_as_current_span("intent_classification") as span:
            intent = await self.classify_intent(user_input)
            span.set_attribute("intent", intent)

        with tracer.start_as_current_span("tool_execution") as span:
            tools_used = []
            for tool_call in self.plan_tools(intent):
                with tracer.start_as_current_span(f"tool_{tool_call.name}"):
                    result = await self.execute_tool(tool_call)
                    tools_used.append(tool_call.name)
            span.set_attribute("tools_used", tools_used)

        with tracer.start_as_current_span("response_generation"):
            response = await self.generate_response()

        return response

---

六、Agent 的互操作性：连接人、Agent 与商业

白皮书最后探讨了 Agent 生态系统的互操作性。

6.1 Agent 与人

人机协作模式：

1. 人在回路（Human-in-the-loop）：

   • Agent 执行前请求确认
   • 关键决策点人工审核
   • 复杂任务分段确认

2. 人在旁路（Human-on-the-loop）：

   • Agent 自主执行
   • 人类监控执行过程
   • 异常情况人工介入

3. 人在指挥（Human-in-command）：

   • 人类设定高层目标
   • Agent 自主规划执行
   • 定期汇报进度

交接设计原则：

class HumanHandoff:
    def __init__(self):
        self.handoff_triggers = [
            self.check_sensitive_operation,
            self.check_confidence_threshold,
            self.check_user_frustration,
            self.check_repeated_failures
        ]

    async def should_handoff(self, context) -> bool:
        """判断是否需要人工介入"""
        for trigger in self.handoff_triggers:
            if await trigger(context):
                return True
        return False

    async def handoff(self, context, reason):
        """执行交接"""
        handoff_context = {
            "conversation_history": context.messages,
            "pending_tasks": context.pending_tasks,
            "reason": reason,
            "suggested_action": await self.suggest_action(context)
        }

        # 通知人工客服
        await self.notify_human_agent(handoff_context)

        return "正在为您转接人工客服，请稍候..."

6.2 Agent 与 Agent

跨 Agent 通信协议：

class AgentCommunicationProtocol:
    """Agent 间通信协议"""

    async def send_message(self, from_agent, to_agent, message_type, payload):
        """发送消息"""
        message = {
            "from": from_agent,
            "to": to_agent,
            "type": message_type,  # request, response, broadcast
            "payload": payload,
            "timestamp": time.time(),
            "message_id": generate_id()
        }

        await self.message_bus.send(message)

    async def request_capability(self, agent_id, capability_requirement):
        """请求其他 Agent 的能力"""
        available_agents = await self.discover_agents(capability_requirement)

        if not available_agents:
            raise NoAgentAvailable(capability_requirement)

        # 选择最佳 Agent
        selected = self.select_best_agent(available_agents)

        # 发送请求
        response = await self.send_request(selected, capability_requirement)
        return response

6.3 Agent 与商业

Agent 经济模型：

1. 按使用付费（Pay-per-use）：

   • 按 API 调用次数计费
   • 按 Token 消耗计费
   • 按任务完成计费

2. 订阅模式（Subscription）：

   • 基础版/专业版/企业版
   • 功能分级
   • 用量配额

3. 效果付费（Outcome-based）：

   • 按业务结果计费
   • 成功才付费
   • 风险共担

---

七、实践启示

7.1 起步建议

渐进式演进路径：

阶段 1（1-2 周）：Level 1 基础 Agent

• 实现单 Agent + 3-5 个基础工具
• 完成一个具体场景（如问答助手）
• 建立基础监控

阶段 2（1 个月）：Level 2 规划 Agent

• 添加任务分解能力
• 实现多步骤任务执行
• 建立评估体系

阶段 3（2-3 个月）：Level 3 多 Agent 系统

• 设计多 Agent 架构
• 实现 Agent 间协作
• 完善运维体系

阶段 4（6 个月+）：Level 4 自进化系统

• 添加学习能力
• 实现自我优化
• 建立反馈闭环

7.2 常见陷阱

陷阱 1：过度工程

• 症状：为简单任务设计复杂的多 Agent 架构
• 解决：从简单开始，按需演进

陷阱 2：忽视评估

• 症状：上线后发现质量不达标
• 解决：建立评估体系，数据驱动迭代

陷阱 3：提示词硬编码

• 症状：提示词散落在代码各处
• 解决：使用提示词管理系统

陷阱 4：缺乏安全考虑

• 症状：Agent 执行了危险操作
• 解决：建立权限控制和审计机制

陷阱 5：忽视用户体验

• 症状：Agent 能力强大但难用
• 解决：持续收集用户反馈

7.3 工具链推荐

类别	工具	用途
开发框架	LangChain, LlamaIndex	Agent 开发
提示词管理	PromptLayer, Weights & Biases	提示词版本管理
评估测试	OpenAI Evals, Promptflow	质量评估
可观测性	LangSmith, Langfuse	监控追踪
部署	Docker, Kubernetes	生产部署

---

八、未来展望

Agent 技术正在快速发展，我们可以期待：

1. 更强大的推理模型：o1 等推理模型将显著提升 Agent 的规划能力
2. 标准化的 Agent 协议：MCP 等协议将成为行业标准
3. 原生 Agent 支持：操作系统和浏览器将原生支持 Agent
4. Agent 市场：将出现类似 App Store 的 Agent 市场
5. 人机协作新模式：Agent 将成为人类的标准”同事”

---

参考与延伸阅读

• 原文：Kaggle Introduction to Agents Whitepaper
• 作者：Alan Blount, Antonio Gulli, Shubham Saboo, Michael Zimmermann, Vladimir Vuskovic
• 发布日期：2025 年 11 月
• 相关项目：LangChain, AutoGPT, MetaGPT

---

本文采用 CC BY-NC-SA 4.0 许可协议。如需转载，请注明出处。