使用 Python 构建高效的 AI 代理系统：从理论到实践

29 3 月, 2026 16点热度 0人点赞 0条评论

引言

在 2026 年的今天，AI 代理（AI Agent）已经成为软件开发和自动化领域的核心概念。从简单的任务自动化到复杂的决策系统，AI 代理正在改变我们与计算机交互的方式。本文将深入探讨如何使用 Python 构建一个高效、可扩展的 AI 代理系统，并提供实用的代码示例，帮助你快速上手。

什么是 AI 代理？

AI 代理是一个能够感知环境、做出决策并执行动作的智能系统。与传统的程序不同，AI 代理具有以下特点：

自主性：能够在没有人类干预的情况下运行
反应性：能够感知环境变化并及时响应
主动性：能够主动追求目标而非被动响应
社交能力：能够与其他代理或人类协作

核心架构设计

1. 代理基础类

让我们从定义一个基础代理类开始：

from abc import ABC, abstractmethod
from typing import Any, Dict, List, Optional
from dataclasses import dataclass
from datetime import datetime
import json

@dataclass
class Message:
    """消息数据结构"""
    role: str  # 'system', 'user', 'assistant'
    content: str
    timestamp: datetime = None
    
    def __post_init__(self):
        if self.timestamp is None:
            self.timestamp = datetime.now()

class BaseAgent(ABC):
    """AI 代理基类"""
    
    def __init__(self, name: str, model: str = "default"):
        self.name = name
        self.model = model
        self.memory: List[Message] = []
        self.state: Dict[str, Any] = {}
    
    @abstractmethod
    def think(self, input: str) -> str:
        """核心思考逻辑"""
        pass
    
    @abstractmethod
    def act(self, decision: str) -> Any:
        """执行动作"""
        pass
    
    def perceive(self, input: str) -> None:
        """感知输入"""
        self.memory.append(Message(role="user", content=input))
    
    def respond(self, output: str) -> None:
        """记录响应"""
        self.memory.append(Message(role="assistant", content=output))
    
    def get_context(self, max_messages: int = 10) -> List[Dict]:
        """获取上下文历史"""
        recent = self.memory[-max_messages:]
        return [{"role": m.role, "content": m.content} for m in recent]

2. 任务代理实现

接下来，我们实现一个具体的任务执行代理，支持异步任务处理和并发控制：

import asyncio
import aiohttp
from enum import Enum

class TaskStatus(Enum):
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"

@dataclass
class Task:
    """任务数据结构"""
    id: str
    description: str
    status: TaskStatus = TaskStatus.PENDING
    result: Any = None
    created_at: datetime = None
    
    def __post_init__(self):
        if self.created_at is None:
            self.created_at = datetime.now()

class TaskAgent(BaseAgent):
    """任务执行代理"""
    
    def __init__(self, name: str, max_concurrent: int = 3):
        super().__init__(name)
        self.tasks: Dict[str, Task] = {}
        self.max_concurrent = max_concurrent
        self.semaphore = asyncio.Semaphore(max_concurrent)
    
    async def create_task(self, description: str) -> Task:
        """创建新任务"""
        task_id = f"task_{len(self.tasks) + 1}"
        task = Task(id=task_id, description=description)
        self.tasks[task_id] = task
        return task
    
    async def execute_task(self, task: Task) -> Any:
        """执行任务（带并发控制）"""
        async with self.semaphore:
            task.status = TaskStatus.RUNNING
            try:
                result = await self._process_task(task)
                task.status = TaskStatus.COMPLETED
                task.result = result
                return result
            except Exception as e:
                task.status = TaskStatus.FAILED
                task.result = str(e)
                raise

3. 多代理协作系统

单个代理能力有限，让我们构建一个多代理协作系统：

class AgentCoordinator:
    """多代理协调器"""
    
    def __init__(self):
        self.agents: Dict[str, BaseAgent] = {}
        self.shared_memory: Dict[str, Any] = {}
    
    def register_agent(self, name: str, agent: BaseAgent) -> None:
        """注册代理"""
        self.agents[name] = agent
    
    async def delegate_task(self, task: str, agent_name: Optional[str] = None) -> Any:
        """分配任务给合适的代理"""
        if agent_name:
            agent = self.agents.get(agent_name)
            if not agent:
                raise ValueError(f"Agent {agent_name} not found")
        else:
            agent = list(self.agents.values())[0]
        
        agent.state["shared"] = self.shared_memory
        
        if isinstance(agent, TaskAgent):
            task_obj = await agent.create_task(task)
            result = await agent.execute_task(task_obj)
        else:
            result = agent.think(task)
        
        self.shared_memory[f"last_result_{agent.name}"] = result
        return result