引言:AI Agent开发范式的演进

2026年,AI Agent的开发已从早期的单一模型调用,演进为模块化、标准化和工具化的系统工程。在这一演进过程中,三种核心范式脱颖而出,共同构成了现代AI Agent开发的基石:

  1. CLI(命令行界面):作为人机交互与自动化执行的基础入口
  2. MCP(模型上下文协议):作为连接AI模型与外部工具/数据的标准化通信层
  3. Skill(技能):作为封装可复用业务逻辑与工具调用的功能单元

本文将深入剖析这三大范式,阐明其各自角色、交互关系,并通过一个完整的“智能数据分析助手”Agent实战项目,展示如何将它们有机结合,构建强大、可维护的AI应用。

范式一:CLI - 统一的控制面与执行入口

CLI是开发者、用户乃至其他系统与AI Agent交互的首要界面。它超越了传统命令执行,成为Agent的“总控台”。

核心价值

  • 标准化入口:为Agent的所有功能(对话、任务执行、技能调用)提供一致的启动和配置方式。
  • 自动化集成:易于嵌入CI/CD流水线、cron作业或其他自动化脚本。
  • 环境与配置管理:集中处理环境变量、认证信息、模型选择等配置。

实战:构建一个Agent CLI骨架

我们使用 argparse (Python标准库) 或 typer (更现代) 来创建CLI。以下使用 typer 为例:

# 首先安装依赖
pip install typer rich
# agent_cli.py
import typer
from typing import Optional
import json
from rich.console import Console
from rich.table import Table
from rich import print as rprint

app = typer.Typer(help="🤖 智能数据分析助手 CLI")
console = Console()

# 模拟的Agent核心和技能库
agent_core = {"status": "ready"}
skills_registry = ["data_fetcher", "chart_generator", "report_writer"]

@app.command()
def start(
    task: str = typer.Argument(..., help="启动Agent执行的具体任务描述"),
    model: str = typer.Option("gpt-4", "--model", "-m", help="选择使用的AI模型"),
    verbose: bool = typer.Option(False, "--verbose", "-v", help="显示详细日志")
):
    """
    启动AI Agent执行一项任务。
    """
    rprint(f"[bold green]🚀 启动Agent...[/bold green]")
    rprint(f"  任务: {task}")
    rprint(f"  模型: {model}")
    rprint(f"  详细模式: {verbose}")
    # 此处会触发后续的MCP调用和Skill执行
    # 例如:process_task_via_mcp(task, model)
    rprint("[bold green]✅ 任务已接收并进入处理队列。[/bold green]")

@app.command()
def list_skills():
    """列出当前Agent已注册的所有技能。"""
    table = Table(title="🛠️ 已注册技能")
    table.add_column("技能名称", style="cyan")
    table.add_column("描述", style="magenta")
    table.add_column("状态", style="green")

    # 模拟从技能库获取信息
    for skill in skills_registry:
        table.add_row(skill, f"执行与'{skill}'相关的操作", "✅ 可用")
    console.print(table)

@app.command()
def status():
    """查看Agent核心状态。"""
    rprint(json.dumps(agent_core, indent=2))

if __name__ == "__main__":
    app()

使用这个CLI:

# 查看帮助
python agent_cli.py --help

# 启动一个任务
python agent_cli.py start "分析本月销售数据并生成趋势图" --model claude-3-5-sonnet

# 列出技能
python agent_cli.py list-skills

# 查看状态
python agent_cli.py status

CLI范式要点:CLI不包含核心业务逻辑,它只是一个调度器配置器,其命令最终会转化为对MCP层的调用。

范式二:MCP - 模型上下文的标准化协议

MCP (Model Context Protocol) 是一种开放协议,用于在AI模型(如ChatGPT、Claude)与工具(Tools)数据源(Data Sources) 之间建立标准化、声明式的连接。它解决了模型如何安全、可控地访问外部能力的问题。

核心概念

  • Server(服务器):提供工具或数据访问能力的后端服务。一个Agent可以连接多个MCP Server。
  • Tools(工具):Server暴露的可调用函数,模型可以决定在何时、传入何种参数来调用它们。
  • Resources(资源):Server提供的可读数据源(如数据库表、文件列表)。
  • Prompts(提示词):可复用的提示词模板。

实战:为Agent实现一个MCP Server(数据获取工具)

假设我们有一个“数据获取”MCP Server,它向AI模型暴露一个 query_sales_data 工具。

# mcp_data_server.py
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.server import Server
from mcp.server.models import Tool
import pandas as pd
from datetime import datetime, timedelta
import json

# 创建MCP Server实例
server = Server("sales-data-server")

# 1. 声明Server提供的工具
@server.list_tools()
async def handle_list_tools() -> list[Tool]:
    return [
        Tool(
            name="query_sales_data",
            description="查询指定日期范围内的销售数据。",
            inputSchema={
                "type": "object",
                "properties": {
                    "start_date": {"type": "string", "description": "开始日期 (YYYY-MM-DD)"},
                    "end_date": {"type": "string", "description": "结束日期 (YYYY-MM-DD)"},
                    "product_category": {"type": "string", "description": "产品类别(可选)", "default": "all"}
                },
                "required": ["start_date", "end_date"]
            }
        )
    ]

# 2. 实现工具的执行逻辑
@server.call_tool()
async def handle_call_tool(name: str, arguments: dict) -> dict:
    if name == "query_sales_data":
        return await execute_query_sales_data(**arguments)
    raise ValueError(f"未知工具: {name}")

async def execute_query_sales_data(start_date: str, end_date: str, product_category: str = "all") -> dict:
    """模拟查询销售数据。"""
    # 在实际应用中,这里会连接数据库
    # 此处返回模拟数据
    dates = pd.date_range(start_date, end_date)
    data = []
    for date in dates:
        for category in ['电子产品', '家居用品', '图书']:
            if product_category != 'all' and category != product_category:
                continue
            revenue = 1000 + (hash(f"{date}{category}") % 500)  # 模拟收入
            data.append({"date": date.strftime("%Y-%m-%d"), "category": category, "revenue": revenue})

    df = pd.DataFrame(data)
    summary = df.groupby('category')['revenue'].sum().to_dict()
    
    return {
        "status": "success",
        "message": f"查询到 {len(df)} 条记录。",
        "data_preview": df.head().to_dict('records'),  # 预览前5条
        "summary": summary,
        "metadata": {
            "query_params": {"start_date": start_date, "end_date": end_date, "product_category": product_category}
        }
    }

# 3. 运行Server(通常通过stdio与模型客户端通信)
async def main():
    async with server.run_stdio_server() as (read_stream, write_stream):
        # 这里Server会持续运行,等待模型客户端的指令
        await asyncio.Future()  # 永久运行

if __name__ == "__main__":
    asyncio.run(main())

MCP范式要点:MCP Server是能力的提供者,它通过标准协议告诉模型“我能做什么”(list_tools),并在模型请求时“执行具体操作”(call_tool)。Agent的核心(或模型)作为MCP Client,会动态发现并调用这些工具。

范式三:Skill - 模块化与可复用的功能单元

Skill是对一个或多个相关工具调用、决策逻辑和提示词工程的高层封装。它代表Agent能完成的一个完整“任务”或“能力”。

核心特征

  • 声明式定义:描述Skill的输入、输出、所需工具和前置条件。
  • 可组合性:简单的Skill可以组合成复杂的Skill(例如,“获取数据” + “生成图表” = “生成数据报告”)。
  • 独立可测试:每个Skill都可以在脱离主Agent的情况下进行单元测试。

实战:定义并实现一个“生成销售图表”Skill

# skills/chart_generator_skill.py
from typing import TypedDict, Optional
import plotly.graph_objects as go
import plotly.io as pio
import json
import os

class ChartInput(TypedDict):
    """生成图表Skill的输入参数。"""
    data: list  # 数据列表,通常来自`query_sales_data`工具的结果
    chart_type: str  # 图表类型,如 'line', 'bar', 'pie'
    title: str
    output_format: str  # 'html', 'png', 'json'

class ChartOutput(TypedDict):
    """生成图表Skill的输出结果。"""
    status: str
    chart_path: Optional[str]  # 生成的图表文件路径
    chart_html: Optional[str]   # 如果是HTML格式,直接返回内容
    error_message: Optional[str]

class ChartGeneratorSkill:
    """技能:根据数据生成可视化图表。"""

    name = "generate_sales_chart"
    description = "将销售数据转换为可视化图表。"
    required_tools = ["query_sales_data"]  # 此技能依赖的工具

    @classmethod
    def execute(cls, input_params: ChartInput, context: dict) -> ChartOutput:
        """
        执行技能。
        :param input_params: 技能输入参数。
        :param context: 执行上下文,包含可用的工具调用函数等。
        :return: 技能执行结果。
        """
        try:
            data = input_params['data']
            chart_type = input_params['chart_type']
            title = input_params['title']

            # 1. 数据转换 (根据实际数据结构调整)
            # 假设data是 `query_sales_data` 返回的 `data_preview` 格式
            df_data = pd.DataFrame(data)
            if df_data.empty:
                return {"status": "error", "error_message": "输入数据为空。"}

            # 2. 根据图表类型创建Plotly图形
            fig = None
            if chart_type == 'line':
                fig = go.Figure()
                for category in df_data['category'].unique():
                    cat_data = df_data[df_data['category'] == category]
                    fig.add_trace(go.Scatter(x=cat_data['date'], y=cat_data['revenue'],
                                             mode='lines+markers', name=category))
                fig.update_layout(title=title, xaxis_title='日期', yaxis_title='销售额')
            
            elif chart_type == 'bar':
                # 按类别聚合
                summary = df_data.groupby('category')['revenue'].sum().reset_index()
                fig = go.Figure([go.Bar(x=summary['category'], y=summary['revenue'])])
                fig.update_layout(title=title, xaxis_title='产品类别', yaxis_title='总销售额')

            else:
                return {"status": "error", "error_message": f"不支持的图表类型: {chart_type}"}

            # 3. 输出处理
            output_format = input_params.get('output_format', 'html')
            output_dir = "output/charts"
            os.makedirs(output_dir, exist_ok=True)
            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            filename = f"sales_chart_{timestamp}"

            if output_format == 'html':
                html_path = os.path.join(output_dir, f"{filename}.html")
                fig.write_html(html_path)
                with open(html_path, 'r', encoding='utf-8') as f:
                    html_content = f.read()
                return {
                    "status": "success",
                    "chart_path": html_path,
                    "chart_html": html_content
                }
            elif output_format == 'png':
                img_path = os.path.join(output_dir, f"{filename}.png")
                fig.write_image(img_path)
                return {"status": "success", "chart_path": img_path}
            else:
                return {"status": "error", "error_message": f"不支持的输出格式: {output_format}"}

        except Exception as e:
            return {"status": "error", "error_message": f"生成图表时出错: {str(e)}"}

# 技能的使用示例
if __name__ == "__main__":
    # 模拟从MCP工具调用获取的数据
    mock_data = [
        {"date": "2026-07-01", "category": "电子产品", "revenue": 1200},
        {"date": "2026-07-01", "category": "家居用品", "revenue": 800},
        {"date": "2026-07-02", "category": "电子产品", "revenue": 1500},
        {"date": "2026-07-02", "category": "家居用品", "revenue": 950},
    ]
    
    skill_input: ChartInput = {
        "data": mock_data,
        "chart_type": "line",
        "title": "近两日销售趋势",
        "output_format": "html"
    }
    
    skill = ChartGeneratorSkill()
    result = skill.execute(skill_input, context={})
    print(json.dumps(result, indent=2, ensure_ascii=False))

Skill范式要点:Skill是业务的封装。它知道完成一个特定任务需要调用哪些工具(通过MCP),如何处理这些工具的返回结果,以及如何将结果转化为最终输出。Skill使Agent的能力变得可插拔、可管理。

综合实战:构建“智能数据分析助手”Agent

现在,我们将三大范式串联起来,构建一个完整的Agent。

项目结构

smart_data_agent/
├── agent_cli.py          # CLI入口 (范式一)
├── mcp_servers/         # MCP Server集合 (范式二)
│   ├── data_server.py   # 数据查询Server
│   └── chart_server.py  # 图表生成Server (可独立,也可作为Skill内部调用)
├── skills/              # Skill集合 (范式三)
│   ├── chart_generator_skill.py
│   └── report_skill.py  # (可扩展)组合数据获取和图表生成的报告Skill
├── agent_core.py        # Agent核心,协调CLI、MCP Client和Skill
└── requirements.txt

Agent核心协调逻辑 (agent_core.py 简化版)

# agent_core.py
import asyncio
from typing import List, Dict, Any
from mcp import ClientSession, StdioServerParameters
import importlib
import sys
import os

class SmartDataAgent:
    def __init__(self):
        self.mcp_clients: Dict[str, ClientSession] = {}
        self.skills_registry: Dict[str, Any] = {}
        self.load_skills()

    def load_skills(self):
        """动态加载skills目录下的所有技能。"""
        skills_dir = "skills"
        for filename in os.listdir(skills_dir):
            if filename.endswith("_skill.py"):
                module_name = f"skills.{filename[:-3]}"
                spec = importlib.util.spec_from_file_location(module_name, os.path.join(skills_dir, filename))
                module = importlib.util.module_from_spec(spec)
                sys.modules[module_name] = module
                spec.loader.exec_module(module)
                # 假设每个skill模块都有一个`SkillClass`类
                for attr_name in dir(module):
                    attr = getattr(module, attr_name)
                    if hasattr(attr, 'name') and hasattr(attr, 'execute'):
                        self.skills_registry[attr.name] = attr
                        print(f"✅ 已加载技能: {attr.name}")

    async def connect_to_mcp_server(self, server_command: List[str]):
        """连接到一个MCP Server。"""
        params = StdioServerParameters(command=server_command[0], args=server_command[1:])
        session = ClientSession(params)
        await session.__aenter__()  # 简化连接
        # 初始化会话,获取Server提供的工具列表
        await session.initialize()
        tools = await session.list_tools()
        server_name = server_command[0]
        self.mcp_clients[server_name] = session
        print(f"🔗 已连接MCP Server: {server_name}, 可用工具: {[t.name for t in tools.tools]}")
        return session

    async def execute_skill(self, skill_name: str, skill_input: Dict, context: Dict) -> Dict:
        """执行一个指定的技能。"""
        if skill_name not in self.skills_registry:
            return {"status": "error", "message": f"技能 '{skill_name}' 未找到。"}
        
        skill_class = self.skills_registry[skill_name]
        # 技能执行可能需要调用MCP工具,这里需要将MCP client传递给技能上下文
        context['mcp_clients'] = self.mcp_clients
        result = skill_class.execute(skill_input, context)
        return result

    async def process_task(self, task_description: st
Logo

中国智能体开发者社区,聚焦智能体与大模型开发,提供前沿资讯、实用工具链、开源项目及行业案例。通过技术沙龙、开发者大赛等活动,促进经验交流与协作,助力开发者快速构建创新智能应用。

更多推荐