---

前言

想象一下特朗普和拜登因为一个热点话题而激情辩论。

一、智能体

1-1、Agent概述

Agent（智能体）： 具有一定自主性和目标导向性，可以在没有持续人类干预的情况下执行任务和作出决策。以下为Agent的一些特性：

（1）自主性和目标导向性

• 自主性：Agent具备自主执行任务的能力，不需要外部指令即可根据设定的目标进行操作。
• 目标导向性：Agent设置并追求特定的目标或任务，这些目标指导其决策过程和行为模式。

（2）复杂的工作流程

• 任务规划与执行：Agent能够规划如何达到其目标，包括任务分解、优先级排序以及实际执行。
• 自我对话和内部决策：在处理问题时，Agent可以进行内部对话，以自我推理和修正其行动路径，而无需外部输入。

（3）学习和适应能力

• 反思和完善：Agent能从自身的经验中学习，评估过去的行为，从错误中吸取教训，并改进未来的策略。
• 环境适应性：在遇到变化的环境或不同的挑战时，Agent能够适应并调整其行为以最大化目标达成。

（4）记忆机制

• 短期记忆：使用上下文信息来做出即时决策。
• 长期记忆：保留关键信息，供未来决策使用，通常通过外部数据库或持久存储实现。（例如使用向量数据库）

（5）工具使用与集成

• API调用和外部数据访问：Agent可以利用外部资源（如API、数据库）来获取信息，填补其知识空白，或执行无法直接通过模型内部处理的任务。
• 技术整合：Agent能整合多种技术和服务，如代码执行能力和专业数据库访问，以丰富其功能和提高效率。

LLM 驱动的自主Agents系统概述如下图所示：（包含工具调用、记忆、计划、执行模块）

1-2、Agent与ChatGPT的区别

Agent与ChatGPT的区别: Agent与ChatGPT在设计、功能和目标上有一些关键区别。虽然它们都是基于人工智能技术，但应用方式和交互性质大不相同。下面是这两者的主要区别：

（1）目标和自主性

• ChatGPT：主要是一个响应型模型，专注于对用户的特定输入生成一次性、相关且连贯的回答。它的主要目的是解答问题、提供信息或进行对话模拟。
• AI Agent：更强调在持续的任务中表现出自主性。它能够设定和追求长期目标，通过复杂的工作流程自主地完成任务，比如从错误中自我修正、连续地追踪任务进展等。

（2） 交互方式

• ChatGPT：用户与ChatGPT的交互通常是线性的和短暂的，即用户提问，ChatGPT回答。它不保留交互的历史记忆，每次交互都是独立的。
• AI Agent：可以维持跨会话的状态和记忆，具有维持长期对话的能力，能够自动执行任务并处理一系列相关活动，例如调用API、追踪和更新状态等。

（3）任务执行和规划能力

• ChatGPT：通常只处理单个请求或任务，依赖用户输入来驱动对话。它不具备自我规划或执行连续任务的能力。
• AI Agent：具备规划能力，可以自行决定执行哪些步骤以完成复杂任务。它可以处理任务序列，自动化决策和执行过程。

（4）技术整合与应用

• ChatGPT：主要是文本生成工具，虽然能够通过插件访问外部信息，但核心依然是文本处理和生成。
• AI Agent：可能整合多种技术和工具，如API调用、数据库访问、代码执行等，这些都是为了实现其目标和改善任务执行的效率。

（5）学习和适应

• ChatGPT：它的训练是在离线进行，通过分析大量数据来改进。
• AI Agent：除了离线学习，更复杂的AI Agent可能具备实时学习能力，能够从新的经验中迅速适应和改进，这通常需要一定的记忆和自我反思机制。

二、多智能体框架MetaGPT（特朗普与拜登的二人辩论）

2-1、安装&配置

安装： 必须要python版本在3.9以上 ，这里使用conda，尝鲜安装。

conda create -n metagpt python=3.9 && conda activate metagpt

开发模式下安装： 为开发人员推荐。实现新想法和定制化功能。

git clone https://github.com/geekan/MetaGPT.git
cd ./MetaGPT
pip install -e .

模型配置： 在文件 ~/.metagpt/config2.yaml下，有关于各大厂商模型的配置详细列表参考：LLM API Configuration

llm:
  api_type: "openai"  # or azure / ollama / groq etc. Check LLMType for more options
  model: "gpt-4-turbo"  # or gpt-3.5-turbo
  base_url: "https://api.openai.com/v1"  # or forward url / other llm url
  api_key: "YOUR_API_KEY"

2-2、使用已有的Agent（ProductManager）

概述： 调用ProductManager Agent，注意，会话上下文是需要独立创建的

import asyncio

from metagpt.context import Context
from metagpt.roles.product_manager import ProductManager
from metagpt.logs import logger

async def main():
    msg = "Write a PRD for a snake game"
    context = Context()  # The session Context object is explicitly created, and the Role object implicitly shares it automatically with its own Action object
    role = ProductManager(context=context)
    while msg:
        msg = await role.run(msg)
        logger.info(str(msg))

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

输出结果：

2-3、智能体对抗辩论——背景

想象一下，如果我们模拟代表拜登和特朗普的智能体共同合作会怎样。这是一个有趣的实验，不是吗？考虑到他们已知的分歧，这样的组合可能导致一些生动的交流。这是一个展示如何设计多个智能体并促进它们之间的互动的理想例子。我们将称呼这个实验为“拜登-特朗普辩论”。

总体上，我们需要3个步骤来设定它们的辩论：

• 定义一个具有发言行为的辩手角色，我们建议参考智能体入门
• 处理辩手之间的通信，也就是让拜登听特朗普说话，反之亦然
• 初始化两个辩手实例，拜登和特朗普，创建一个带有环境的团队，并使它们能够相互交互

2-4、定义动作

这里我们简单修改一下官方的提示词模板，官方提示词模板如下所示：

PROMPT_TEMPLATE = """
    ## BACKGROUND
    Suppose you are {name}, you are in a debate with {opponent_name}.
    ## DEBATE HISTORY
    Previous rounds:
    {context}
    ## YOUR TURN
    Now it's your turn, you should closely respond to your opponent's latest argument, state your position, defend your arguments, and attack your opponent's arguments,
    craft a strong and emotional response in 80 words, in {name}'s rhetoric and viewpoints, your will argue:
    """

SpeakAloud 动作定义如下：

• 简单修改下提示词模板，英文翻译为中文，可以加一些要求，比如用中文回复，回复字数是多少，回复应该言辞激烈等等。
• 这里一定要参考背景，即上下文context。

import asyncio
import platform
from typing import Any

import fire

from metagpt.actions import Action, UserRequirement
from metagpt.logs import logger
from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.team import Team

class SpeakAloud(Action):
    """动作：在辩论中大声说话（争吵）"""

    PROMPT_TEMPLATE = """
    ## 背景
    假设你是 {name}, 你正在和 {opponent_name} 辩论.
    ## 辩论过程
    之前的回合:
    {context}
    ## 轮到你
    现在轮到你了，你应该密切回应对手的最新论点，陈述你的立场，为你的论点辩护，攻击你对手的论点，
    用100个字写出一个强烈而情绪化的回应，用中文回复, 以 {name} 的修辞和观点, 你将要发言:
    """

    def __init__(self, name="SpeakAloud", context=None, llm=None):
        super().__init__(name, context, llm)

    async def run(self, context: str, name: str, opponent_name: str):

        prompt = self.PROMPT_TEMPLATE.format(context=context, name=name, opponent_name=opponent_name)

        rsp = await self._aask(prompt)

        return rsp

2-5、定义角色

角色定义：

• name代表名字，opponent_name代表对手的名字，profile代表人物简介，我们可以写一些对于人物的简单介绍
• set_actions定义角色拥有的动作
• _watch 监视用户输入以及 其他角色发起的SpeakAloud动作。我们希望辩手可以密切关注其他角色的发言
• _observe，处理来自对手的论点，而不是自己的。
• _act，将辩论信息发送给对手

class Debator(Role):
    def __init__(
        self,
        name: str,
        profile: str,
        opponent_name: str,
        **kwargs,
    ):
        super().__init__(name, profile, **kwargs)
        self.set_actions([SpeakAloud])
        self._watch([UserRequirement, SpeakAloud])
        self.name = name
        self.opponent_name = opponent_name

    async def _observe(self) -> int:
        await super()._observe()
        # accept messages sent (from opponent) to self, disregard own messages from the last round
        self.rc.news = [msg for msg in self.rc.news if msg.send_to == self.name]
        return len(self.rc.news)

    async def _act(self) -> Message:
        logger.info(f"{self._setting}: ready to {self.rc.todo}")
        todo = self.rc.todo # 一个 SpeakAloud 的实例

        memories = self.get_memories()
        context = "\n".join(f"{msg.sent_from}: {msg.content}" for msg in memories)

        rsp = await todo.run(context=context, name=self.name, opponent_name=self.opponent_name)

        msg = Message(
            content=rsp,
            role=self.profile,
            cause_by=todo,
            sent_from=self.name,
            send_to=self.opponent_name,
        )

        return msg

2-6、创建团队，添加角色

创建团队，添加角色：

• 分别建立角色，拜登和特朗普。身份分别为民主党、共和党党员。
• 让拜登先说话。

async def debate(idea: str, investment: float = 3.0, n_round: int = 5):
    """运行拜登-特朗普辩论，观看他们之间的友好对话 :) """
    Biden = Debator(name="Biden", profile="Democrat", opponent_name="Trump")
    Trump = Debator(name="Trump", profile="Republican", opponent_name="Biden")
    team = Team()
    team.hire([Biden, Trump])
    team.invest(investment)
    team.run_project(idea, send_to="Biden")  # 将辩论主题发送给拜登，让他先说话
    await team.run(n_round=n_round)


import asyncio
import platform
import typer
from metagpt.team import Team
app = typer.Typer()

@app.command()
def main(
    idea: str = typer.Argument(..., help="Economic Policy: Discuss strategies and plans related to taxation, employment, fiscal budgeting, and economic growth."),
    investment: float = typer.Option(default=3.0, help="Dollar amount to invest in the AI company."),
    n_round: int = typer.Option(default=5, help="Number of rounds for the simulation."),
):
    """
    :param idea: Debate topic, such as "Topic: The U.S. should commit more in climate change fighting"
                 or "Trump: Climate change is a hoax"
    :param investment: contribute a certain dollar amount to watch the debate
    :param n_round: maximum rounds of the debate
    :return:
    """
    if platform.system() == "Windows":
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
    asyncio.run(debate(idea, investment, n_round))

if __name__ == '__main__':
    app()

输出： 论题（讨论一下日益增长的中国威胁）

三、构建StreamLit界面

3-1、智能体脚本

智能体脚本： 命名为debate_with_streamLit.py，用到的是上边的案例。主要需要修改的是debate函数

• 每一轮对话之后使用history获取到历史。
• 使用正则表达式分别匹配到特朗普以及拜登的发言
• 使用flag当作标记去依次添加特朗普和拜登的发言。
• 最后把整个对话记录返回，在streamlit界面接收记录并且进行展示。

import asyncio
import platform
from typing import Any

from metagpt.actions import Action, UserRequirement
from metagpt.logs import logger
from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.team import Team
import re


class SpeakAloud(Action):
    """Action: Speak out aloud in a debate (quarrel)"""

    PROMPT_TEMPLATE: str = """
    ## 背景
    假设你是 {name}, 你正在和 {opponent_name} 辩论.
    ## 辩论过程
    之前的回合:
    {context}
    ## 轮到你
    现在轮到你了，你应该密切回应对手的最新论点，陈述你的立场，为你的论点辩护，攻击你对手的论点，
    用100个字写出一个强烈而情绪化的回应，用中文回复, 以 {name} 的修辞和观点, 你将要发言:
    """
    name: str = "SpeakAloud"

    async def run(self, context: str, name: str, opponent_name: str):
        prompt = self.PROMPT_TEMPLATE.format(context=context, name=name, opponent_name=opponent_name)
        rsp = await self._aask(prompt)
        return rsp


class Debator(Role):
    name: str = ""
    profile: str = ""
    opponent_name: str = ""

    def __init__(self, **data: Any):
        super().__init__(**data)
        self.set_actions([SpeakAloud])
        self._watch([UserRequirement, SpeakAloud])

    async def _observe(self) -> int:
        await super()._observe()
        # accept messages sent (from opponent) to self, disregard own messages from the last round
        self.rc.news = [msg for msg in self.rc.news if msg.send_to == {self.name}]
        return len(self.rc.news)

    async def _act(self) -> Message:
        logger.info(f"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})")
        todo = self.rc.todo  # An instance of SpeakAloud

        memories = self.get_memories()
        context = "\n".join(f"{msg.sent_from}: {msg.content}" for msg in memories)

        rsp = await todo.run(context=context, name=self.name, opponent_name=self.opponent_name)

        msg = Message(
            content=rsp,
            role=self.profile,
            cause_by=type(todo),
            sent_from=self.name,
            send_to=self.opponent_name,
        )
        self.rc.memory.add(msg)

        return msg


async def debate(idea: str, investment: float = 3.0, n_round: int = 5):
    """Run a team of presidents and watch them quarrel."""

    Biden = Debator(name="Biden", profile="Democrat", opponent_name="Trump")
    Trump = Debator(name="Trump", profile="Republican", opponent_name="Biden")
    team = Team()
    team.hire([Biden, Trump])
    team.invest(investment)
    team.run_project(idea, send_to="Biden")

    results = []
    # Ensure this collects messages properly
    flag = True
    for _ in range(n_round):
        await team.run(n_round=1)  # Run each round separately

        text = team.env.history
        pattern_Republican = r"Republican[^:]*: ([^\n]+)"
        pattern_Democrat = r"Democrat[^:]*: ([^\n]+)"
        match_Republican = re.findall(pattern_Republican, text, re.DOTALL)
        match_Democrat = re.findall(pattern_Democrat, text, re.DOTALL)
        if flag:
            # Extract and print the last statement
            last_statement = '拜登：' + match_Democrat[-1].strip()
            flag = False
        else:
            last_statement = '特朗普：' + match_Republican[-1].strip()
            flag = True

        results.append(last_statement)  # Collect all messages into results
    return results


def run_debate_from_streamlit(idea: str, investment: float = 3.0, n_round: int = 5):
    """
    :param idea: Debate topic
    :param investment: Initial investment amount
    :param n_round: Maximum rounds of the debate
    :return: Debate results
    """
    if platform.system() == "Windows":
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
    results = asyncio.run(debate(idea, investment, n_round))
    return results


if __name__ == "__main__":
    idea = "讨论一下日益增长的中国威胁"
    results = run_debate_from_streamlit(idea=idea)
    # for result in results:
    #     print(f"{result.sent_from}: {result.content}")  # Print content of the messages

3-2、streamlit页面构建

启动：

streamlit run streamlit_debate.py

streamlit页面构建代码如下： 这里智能体代码命名为debate_with_streamLit，streamlit需要调用智能体

import streamlit as st
from debate_with_streamLit import run_debate_from_streamlit

def main():
    st.title("Debate Simulation")

    # Input fields for Streamlit
    idea = st.text_input("Enter the debate topic:", "讨论一下日益增长的中国威胁")
    investment = st.number_input("Enter the investment amount:", min_value=0.0, value=3.0, step=0.1)
    n_round = st.number_input("Enter the number of debate rounds:", min_value=1, max_value=20, value=5, step=1)

    if st.button("Start Debate"):
        st.write("Debate starting...")
        with st.spinner("Running debate..."):
            # Ensure run_debate_from_streamlit returns a list of Message objects
            results = run_debate_from_streamlit(idea, investment, n_round)
        st.success("Debate complete!")
        if results:
            # message_container = st.empty()  # Container for updating messages
            # for round_number, msg in enumerate(results, start=1):
            #     message_container.write(f"**Round {round_number}:** **{msg.sent_from}:** {msg.content}")
            #     st.time.sleep(1)  # Simulate a delay between messages
            st.write(results)
        else:
            st.write("No results returned from debate.")


if __name__ == "__main__":
    main()

界面展示如下：

总结（优化空间）：
1、一次性回复所有，等待时间较长，后期可以流式回复
2、streamlit界面不太友好，有优化空间
3、响应较慢，可以进一步去提升速度

四、多智能体实践（数字化永生问题——四智能体的辩论赛）

对官方案例做一些修改，做一个四智能体的辩论赛。

4-1、定义动作

SpeakAloud 动作定义如下：

• 简单修改下提示词模板，为了让角色对个人定位更加准确，我们引入了profile，即个人信息。
• 因为这里是四位角色，所以我们不仅要给出队友，还要给出对手。
• 第一位发言的选手，我们需要让他去介绍辩论主题，@一下队友和对手，最后表达自己的观点
• 如果不是第一位发言的选手，我们需要让他关注前一个人说的话。

import asyncio
import platform
from typing import Any

import fire

from metagpt.actions import Action, UserRequirement
from metagpt.logs import logger
from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.team import Team
import random


class SpeakAloud(Action):
    """Action: Speak out aloud in a debate (quarrel)"""

    PROMPT_TEMPLATE: str = """
    ## 背景
    你是 {name}, 有关于你的一些个人信息是：{profile},你正在参加一场辩论赛，辩论主题为：{idea},
    你的队友是{friend_name}，你的对手是{opponent_name}.
    ## 辩论过程
    之前的回合:
    {context}
    ## 轮到你
    现在轮到你了，
    如果没有之前的回合，则你需要简明扼要的介绍辩论主题，让观众、队友、对手清楚此次辩论的核心议题。并且清晰的表达自己的立场或观点。
    如果上一句话是你的对手说的，你应该密切回应对手的最新论点，陈述你的立场，为你的论点辩护，攻击你对手的论点，
    如果上一句话是你的队友说的，你应该赞成队友的观点，为队友的观点辩护，
    用200个字写出一个强烈而情绪化的回应，用中文回复, 以 {name} 的修辞和观点, 
    你将要发言:
    """
    name: str = "SpeakAloud"

    async def run(self, context: str, name: str, opponent_name: str, friend_name: str, profile: str, idea: str):
        prompt = self.PROMPT_TEMPLATE.format(context=context, name=name, opponent_name=opponent_name,
                                             friend_name=friend_name, profile=profile, idea=idea)

        rsp = await self._aask(prompt)

        return rsp

4-2、定义角色

角色定义：

• 相较于官方案例，基本不需要更改什么，但是需要添加friend_name、idea参数
• name代表名字，opponent_name代表对手的名字，profile代表人物简介，我们可以写一些对于人物的简单介绍
• set_actions定义角色拥有的动作
• _watch 监视用户输入以及 其他角色发起的SpeakAloud动作。我们希望辩手可以密切关注其他角色的发言
• _observe，处理来自对手的论点，而不是自己的。
• _act，将辩论信息发送给对手
• 在执行动作todo.run时，需要把新增的参数填入
• 定义Message函数时，需要简单修改下send_to参数，这里的设定为随机选择一个队友或者对手发言，这样更真实

class Debator(Role):
    name: str = ""
    profile: str = ""
    opponent_name: list = []
    friend_name: str = ""
    idea: str = ""

    def __init__(self, **data: Any):
        super().__init__(**data)
        self.set_actions([SpeakAloud])
        self._watch([UserRequirement, SpeakAloud])

    async def _observe(self) -> int:
        await super()._observe()
        # accept messages sent (from opponent) to self, disregard own messages from the last round
        self.rc.news = [msg for msg in self.rc.news if msg.send_to == {self.name}]
        return len(self.rc.news)

    async def _act(self) -> Message:
        logger.info(f"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})")
        todo = self.rc.todo  # An instance of SpeakAloud

        memories = self.get_memories()
        context = "\n".join(f"{msg.sent_from}: {msg.content}" for msg in memories)
        # print(context)

        rsp = await todo.run(context=context, name=self.name, opponent_name=self.opponent_name,
                             friend_name=self.friend_name, profile=self.profile, idea=self.idea)

        msg = Message(
            content=rsp,
            role=self.profile,
            cause_by=type(todo),
            sent_from=self.name,
            send_to=random.choice(self.opponent_name+[self.friend_name])
        )
        self.rc.memory.add(msg)

        return msg

2-6、创建团队，添加角色

创建团队，添加角色：

• 分别建立角色，
• 对角色赋予背景信息，对手名字、朋友名字以及议题是什么

async def debate(idea: str, investment: float = 3.0, n_round: int = 5):
    """Run a team of presidents and watch they quarrel. :)"""

    # profile： 配置文件
    # Biden = Debator(name="Biden", profile="Democrat", opponent_name="Trump")
    # Trump = Debator(name="Trump", profile="Republican", opponent_name="Biden")

    Alan_Smith = Debator(name="Alan_Smith", profile="背景：一家知名人工智能和生物科技公司的创始人兼CEO。"
                                                    "观点：相信科技可以解决人类面临的许多挑战，数字化永生是未来发展的重要方向。",
                         opponent_name=["James_Roberts", "Mary_Gonzalez"], friend_name="Lisa_Chen", idea=idea)
    Lisa_Chen = Debator(name="Lisa_Chen", profile="背景：著名未来学家和科技畅销书作者，专注于人类未来的发展趋势。"
                                                  "观点：数字化永生将改变人类对生命的理解，开辟新的人生体验和发展路径。",
                        opponent_name=["James_Roberts", "Mary_Gonzalez"], friend_name="Alan_Smith", idea=idea)

    James_Roberts = Debator(name="James_Roberts", profile="背景：伦理学教授，专注于科技伦理与人类价值研究。"
                                                          "观点：担忧数字化永生带来的伦理困境和人类本质的丧失。",
                            opponent_name=["Lisa_Chen", "Alan_Smith"], friend_name="Mary_Gonzalez", idea=idea)
    Mary_Gonzalez = Debator(name="Mary_Gonzalez", profile="背景：社会学教授，研究科技对社会结构和人类关系的影响。"
                                                          "观点：认为数字化永生可能导致社会不平等和伦理冲突。",
                            opponent_name=["Lisa_Chen", "Alan_Smith"], friend_name="James_Roberts", idea=idea)

    team = Team()
    team.hire([Alan_Smith, Lisa_Chen, James_Roberts, Mary_Gonzalez])
    team.invest(investment)
    team.run_project(idea, send_to="Alan_Smith")  # send debate topic to Biden and let him speak first
    await team.run(n_round=n_round)


def main(idea: str, investment: float = 3.0, n_round: int = 10):
    """
    :param idea: Debate topic, such as "Topic: The U.S. should commit more in climate change fighting"
                 or "Trump: Climate change is a hoax"
    :param investment: contribute a certain dollar amount to watch the debate
    :param n_round: maximum rounds of the debate
    :return:
    """
    if platform.system() == "Windows":
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
    asyncio.run(debate(idea, investment, n_round))


if __name__ == "__main__":
    fire.Fire(main)  # run as python debate.py --idea="TOPIC" --investment=3.0 --n_round=5

输出： 论题（执行命令：python debate.py --idea “讨论一下数字化永生问题”）

参考文章：

《MetaGPT智能体开发入门》教程
Datawhale教程.
MetaGPT—GitHub官网
openAI研究主管文章
awesome-ai-agents——AI agent汇总
MetaGPT智能体入门——官方文档

LLM图形化界面：
川虎 Chat 🐯 Chuanhu Chat
chatgpt-KnowledgeBot

---

总结

太酷啦。😎