AI 应用与智能体开发：从模型调用到多智能体协作

from abc import ABC, abstractmethod
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Callable

@dataclass
class Message:
    sender: str
    receiver: str
    content: str

class MessageBus:
    def __init__(self):
        self.queues = defaultdict(list)
        self.subscribers = defaultdict(list)

    def register(self, agent_name: str, handler: Callable):
        self.subscribers[agent_name].append(handler)

    def send(self, msg: Message):
        self.queues[msg.receiver].append(msg)
        for handler in self.subscribers[msg.receiver]:
            handler(msg)

class BaseAgent(ABC):
    def __init__(self, name: str, bus: MessageBus):
        self.name = name
        self.bus = bus
        bus.register(name, self.receive)

    def send(self, receiver: str, content: str):
        self.bus.send(Message(self.name, receiver, content))

    @abstractmethod
    def receive(self, msg: Message):
        pass

class Researcher(BaseAgent):
    def receive(self, msg: Message):
        topic = msg.content
        print(f"[研究员] 收到主题：{topic}，开始收集资料...")
        findings = f"关于{topic}的关键事实：1. 定义；2. 现状；3. 趋势。"
        self.send("writer", findings)

class Writer(BaseAgent):
    def receive(self, msg: Message):
        findings = msg.content
        print(f"[写作者] 收到资料，开始撰写...")
        article = f"基于以下资料成文：{findings}\n\n——文章完成。"
        print(article)

# 运行
bus = MessageBus()
researcher = Researcher("researcher", bus)
writer = Writer("writer", bus)

bus.send(Message("user", "researcher", "AI 智能体发展趋势"))

class Tool:
    def __init__(self, name: str, func: Callable, description: str):
        self.name = name
        self.func = func
        self.description = description

class ToolRegistry:
    def __init__(self):
        self.tools = {}

    def register(self, tool: Tool):
        self.tools[tool.name] = tool

    def call(self, name: str, **kwargs):
        if name not in self.tools:
            raise ValueError(f"未知工具: {name}")
        return self.tools[name].func(**kwargs)

registry = ToolRegistry()
registry.register(Tool("search", lambda q: f"搜索'{q}'的结果", "搜索网络信息"))
print(registry.call("search", q="AI Agent"))