Agent World 入门指南：从零构建你的第一个智能体系统

1次阅读

没有评论

共计 2806 个字符，预计需要花费 8 分钟才能阅读完成。

传统的集中式系统在处理复杂场景时，常常会遇到瓶颈。比如一个电商平台的订单处理系统，随着用户量增长，系统会变得越来越臃肿，任何一个小改动都可能引发连锁反应。这种架构主要有三个痛点：

单点故障风险高
扩展性差
响应速度慢

而智能体系统 (Agent System) 通过将功能拆分成多个自治的智能体(Agent)，每个智能体只负责特定功能，相互之间通过消息传递协作。这种架构正好解决了上述问题：

容错性强：单个智能体故障不会影响整个系统
扩展方便：可以动态增减智能体数量
响应快速：智能体可以并行处理任务

在开始编码前，我们需要明确几个关键术语：

Agent（智能体）：具有自主行为能力的实体，能够感知环境并做出决策
Environment（环境）：智能体生存和活动的虚拟空间
Message（消息）：智能体间通信的基本单位

智能体间的通信机制可以用这个简单图示表示：

[Agent A] --Message--> [Agent B]
       \               /
        \             /
         [Environment]

下面我们来实现一个基础智能体类，它具备最基本的功能：感知环境、处理消息和简单决策。

import asyncio
from dataclasses import dataclass
from typing import Any, Dict, List

@dataclass
class Message:
    sender: str  # 发送者 ID
    receiver: str  # 接收者 ID
    content: Any  # 消息内容

class BasicAgent:
    def __init__(self, agent_id: str):
        self.id = agent_id  # 智能体唯一标识
        self.mailbox: List[Message] = []  # 消息收件箱
        self.env = None  # 环境引用

    async def perceive(self):
        """感知环境状态"""
        # 这里可以添加感知逻辑
        pass

    async def process_message(self, msg: Message):
        """处理接收到的消息"""
        print(f"Agent {self.id} received: {msg.content} from {msg.sender}")

        # 简单回复逻辑
        if "hello" in str(msg.content).lower():
            reply = Message(
                sender=self.id,
                receiver=msg.sender,
                content=f"Hi {msg.sender}! I'm {self.id}."
            )
            await self.env.deliver_message(reply)

    async def run(self):
        """智能体主循环"""
        while True:
            await self.perceive()

            # 处理所有待处理消息
            while self.mailbox:
                msg = self.mailbox.pop(0)
                await self.process_message(msg)

            await asyncio.sleep(0.1)  # 防止 CPU 占用过高

class Environment:
    def __init__(self):
        self.agents: Dict[str, BasicAgent] = {}

    def register_agent(self, agent: BasicAgent):
        """注册智能体到环境"""
        agent.env = self
        self.agents[agent.id] = agent

    async def deliver_message(self, msg: Message):
        """投递消息到目标智能体"""
        if msg.receiver in self.agents:
            self.agents[msg.receiver].mailbox.append(msg)

async def main():
    # 创建环境和智能体
    env = Environment()
    alice = BasicAgent("Alice")
    bob = BasicAgent("Bob")

    # 注册智能体
    env.register_agent(alice)
    env.register_agent(bob)

    # 启动智能体
    tasks = [asyncio.create_task(alice.run()),
        asyncio.create_task(bob.run())
    ]

    # 发送初始消息
    await env.deliver_message(Message(
        sender="Alice",
        receiver="Bob",
        content="Hello Bob!"
    ))

    # 运行 10 秒
    await asyncio.sleep(10)

    # 取消任务
    for task in tasks:
        task.cancel()

def run_example():
    asyncio.run(main())

if __name__ == "__main__":
    run_example()

完整代码可以在这个 GitHub 仓库找到。这个示例展示了智能体系统最基本的三个能力：