分布式人工智能与 Agent

"""家用机器人 BDI 智能体：感知 -> 信念 -> 愿望 -> 意图 -> 行动"""

class BDIAgent:
    def __init__(self, name):
        self.name = name
        self.beliefs = set()                 # 信念
        self.intentions = []                 # 意图栈
        # 规划库：意图 -> 动作序列
        self.plans = {
            "充电": ["连接充电桩", "等待电量恢复"],
            "带伞上班": ["检查雨伞", "出发去公司"],
            "公园散步": ["换上运动鞋", "出门散步"],
            "室内看书": ["取一本书", "在沙发阅读"],
        }

    # ① 信念修正：用传感器信息更新信念
    def brf(self, env):
        self.beliefs.clear()
        self.beliefs.add("下雨" if env["weather"] == "雨" else "天晴")
        self.beliefs.add("电量低" if env["battery"] < 30 else "电量充足")

    # ② 选项生成：由信念产生候选愿望
    def options(self):
        desires = []
        if "电量低" in self.beliefs:
            desires.append("充电")                 # 高优先
        if "下雨" in self.beliefs:
            desires += ["带伞上班", "室内看书"]
        if "天晴" in self.beliefs and "电量低" not in self.beliefs:
            desires += ["公园散步"]
        return desires

    # ③ 过滤：从愿望中承诺一个意图（此处取优先级最高者）
    def filter_intention(self, desires):
        self.intentions = [desires[0]] if desires else []

    # ④ 执行意图对应的规划
    def execute(self):
        if not self.intentions:
            print("  无可行意图，保持待机")
            return
        goal = self.intentions[0]
        print(f"  承诺意图：{goal}")
        for act in self.plans[goal]:
            print(f"  执行动作：{act}")

    def step(self, day, env):
        print(f"== 第 {day} 天，环境：{env}")
        self.brf(env)
        print(f"  信念：{sorted(self.beliefs)}")
        desires = self.options()
        print(f"  候选愿望：{desires}")
        self.filter_intention(desires)
        self.execute()

if __name__ == "__main__":
    robot = BDIAgent("家务机器人")
    week = [
        {"weather": "雨", "battery": 80},
        {"weather": "晴", "battery": 80},
        {"weather": "晴", "battery": 15},
    ]
    for day, env in enumerate(week, 1):
        robot.step(day, env)

"""FIPA ACL 风格的智能体消息与对话模拟"""

class ACLMessage:
    def __init__(self, sender, receiver, performative, content, reply_to=None):
        self.sender = sender
        self.receiver = receiver
        self.performative = performative
        self.content = content
        self.reply_to = reply_to

    def __repr__(self):
        return (f"[{self.performative}] {self.sender} -> {self.receiver}："
                f"{self.content}")

class MessageSystem:
    def __init__(self):
        self.boxes = {}
    def register(self, name):
        self.boxes[name] = []
    def send(self, msg):
        self.boxes[msg.receiver].append(msg)

class Agent:
    def __init__(self, name, system):
        self.name, self.system = name, system
        system.register(name)

    def send(self, to, perf, content, reply_to=None):
        self.system.send(ACLMessage(self.name, to, perf, content, reply_to))

    def run(self):
        for msg in list(self.system.boxes[self.name]):
            self.system.boxes[self.name].remove(msg)
            self.handle(msg)

    def handle(self, msg):
        print(msg)

class WeatherAgent(Agent):
    def handle(self, msg):
        print(msg)
        if msg.performative == "QUERY-IF":
            self.send(msg.sender, "INFORM", "今日晴转多云，22~28℃", msg)

class PlannerAgent(Agent):
    def handle(self, msg):
        print(msg)
        if msg.performative == "INFORM":
            print(f"  （{self.name} 收到天气，决定：适宜外出，已更新日程）")

if __name__ == "__main__":
    sys_ = MessageSystem()
    planner = PlannerAgent("规划智能体", sys_)
    weather = WeatherAgent("天气智能体", sys_)

    planner.send("天气智能体", "QUERY-IF", "今天会下雨吗？")
    weather.run()      # 天气智能体回答
    planner.run()      # 规划智能体接收并处理

"""合同网协议模拟：一项运输任务的招标、投标、授标与执行"""

class Task:
    def __init__(self, name, weight, destination):
        self.name, self.weight, self.destination = name, weight, destination

class Bidder:
    def __init__(self, name, capacity, price, days):
        self.name, self.capacity = name, capacity
        self.price, self.days = price, days

    def bid(self, task):
        if self.capacity < task.weight:
            return None                        # 能力不足，放弃投标
        return {"bidder": self, "price": self.price, "days": self.days}

def contract_net(manager, task, bidders):
    print(f"【招标公告】{manager}：{task.name}，"
          f"{task.weight}吨货物运至{task.destination}，请有意者投标")
    proposals = []
    for b in bidders:
        p = b.bid(task)
        if p:
            print(f"【投标】{b.name}：报价 {b.price} 元，{b.days} 天送达")
            proposals.append(p)
        else:
            print(f"【弃标】{b.name}：运力 {b.capacity} 吨，无法承运")

    winner = min(proposals, key=lambda p: p["price"])   # 最低价评标
    print(f"【授标】{manager} 决定由 {winner['bidder'].name} 中标")
    print(f"【执行】{winner['bidder'].name} 运输中…… "
          f"{winner['days']} 天后货物送达，合同完成 ✔")
    return winner

if __name__ == "__main__":
    task = Task("3 吨电子元件运输", 3, "上海")
    bidders = [
        Bidder("迅捷物流", capacity=2, price=4000, days=1),
        Bidder("顺路车队", capacity=5, price=6000, days=3),
        Bidder("安心快运", capacity=8, price=8500, days=2),
    ]
    contract_net("供应链调度中心", task, bidders)

"""多智能体协作博弈：最优反应动态与分工涌现"""

# 联合奖励矩阵 R(智能体1的动作, 智能体2的动作)
PAYOFF = {
    ("采集", "采集"): 1,    # 都采集，无人运输
    ("运输", "采集"): 10,   # 分工：1采集 2运输
    ("采集", "运输"): 10,   # 分工
    ("运输", "运输"): 1,    # 都运输，无人采集
}
ACTIONS = ["采集", "运输"]

def best_response(who, other_action):
    """给定对方动作，选择使联合奖励最大的自身动作。"""
    if who == 1:
        return max(ACTIONS, key=lambda a: PAYOFF[(a, other_action)])
    return max(ACTIONS, key=lambda a: PAYOFF[(other_action, a)])

if __name__ == "__main__":
    a1 = a2 = "采集"                 # 初始策略
    print("轮次  机器人1  机器人2  团队奖励")
    for rnd in range(4):
        reward = PAYOFF[(a1, a2)]
        print(f" {rnd}     {a1}     {a2}      {reward}")
        new_a1 = best_response(1, a2)   # 机器人1 先调整
        new_a2 = best_response(2, new_a1)  # 机器人2 随后适应
        if (new_a1, new_a2) == (a1, a2):
            print("策略不再变化 -> 系统达到均衡，分工稳定")
            break
        a1, a2 = new_a1, new_a2

"""ReAct 智能体骨架（工具调用以桩函数模拟，可替换为真实 LLM 与 API）"""

def search_tool(query):
    # 真实场景中调用搜索引擎 API
    return f"搜索结果：{query} -> 今日晴，气温 22~28℃"

def calculator(expr):
    return str(eval(expr))                    # 教学演示，生产环境需沙箱解析

TOOLS = {"search": search_tool, "calc": calculator}

def react_agent(question, max_steps=4):
    scratchpad = []                          # 思考-行动-观察的工作记忆
    print("用户问题：", question)
    for step in range(max_steps):
        # 规则模拟 LLM 的决策（真实实现中由大模型输出）
        if step == 0:
            thought, action, arg = "需要先查到天气", "search", "今天天气"
        elif step == 1:
            thought, action, arg = "信息已足够，可以作答", "finish", "今天天气晴，适合外出"
        else:
            break
        print(f"思考{step+1}：{thought}")
        if action == "finish":
            print("最终回答：", arg)
            return arg
        observation = TOOLS[action](arg)
        print(f"行动：{action}({arg!r}) -> 观察：{observation}")
        scratchpad.append(observation)

if __name__ == "__main__":
    react_agent("今天适合出门跑步吗？")