CAMEL Multi-Agent 多智能体学习2

一、智能体概述

智能体是一个能够感知环境并在环境中自主行动以实现特定目标的系统。

关键特征：自主性、响应性、主动性、社交能力

智能体的分类：任务型、学习型、协作型、对话型

依赖技术：LLM（认知和决策）

​ 规划系统

​ 记忆系统

​ 工具调用

发展时间线：

  • CAMEL:http://www.camel-ai.org/

  • AutoGPT

  • HuggingGPT

  • Westworld:https://github.com/joonspk-research/generative_agents

  • VOYAGER:http://github.com/MineDojo/Voyager

  • MetaGPT:https://github.com/geekan/MetaGPT

  • ChatDev

  • XAgent:http://github.com/OpenBMB/XAgent

  • Amazon Bedrock Agents:https://aws.amazon.com/bedrock/agents/

  • OpenAssistant Pythia:https://github.com/LAION-AI/Open-Assistant

  • Ray:https://ray-project.github.io/q4-2021-docs-hackathon/0.4/ray-api-references/ray-rllib/agents/

  • OpenAI Swarm:https://github.com/openai/swarm

二、Agent设计原则与方法

原则

  • 目标导向性:智能体的每个行为都应该服务于预定的目标

  • 模块化：将不同功能划分为独立的模块

智能体的核心部分：

  • 感知模块：处理输入信息
  • 决策模块：制定行动计划
  • 执行模块：实施具体行动

同时要注意的是智能体的设计应该采用迭代开发的方式，循序渐进，先实现基本功能，然后通过不断地测试和反馈来完善系统

三、Model及调用

Model 是 Agent 的大脑，负责处理所有输入和输出数据。通过有效调用不同的模型，智能体可以根据任务需求执行文本分析、图像识别和复杂推理等操作

我们可以通过使用CAMEL的ModelFactory的create方法创建不同的model，然后修改以下三个参数就可以做到调用不同的模型：model_platform、model_type、model_config_dict

比如以下：

model = ModelFactory.create(
    model_platform=ModelPlatformType.ZHIPU, #指定了模型所在的平台为智谱
    model_type=ModelType.GLM_4, #选择模型类型为GLM_4
	model_config_dict=ZhipuAIConfig(temperature=0.2).as_dict(), #模型配置
    api_key=os.environ.get("ZHIPUAI_API_KEY"),
    url=os.environ.get("ZHIPUAI_API_BASE_URL"),
)

model_config_dict:这里使用 ZhipuAIConfig 来配置模型的参数：

  • temperature=0.2 是控制生成内容的随机性，较低的温度（如 0.2）会使模型生成的回答更加确定和专注。
  • .as_dict() 方法将配置转换为字典格式，便于传递给模型创建方法。

我们可以稍微改动参数就可以连接到openai的API接口(略)

我们也可以使用开源模型，比如Ollama(提供了一个简单的 API，能够将预训练的语言模型集成到你的应用程序中),或者使用vLLM(面向大规模语言模型的推理库，专为大规模语言模型设计，旨在提高推理速度和减少资源消耗)

四、Agent的Messages
  1. 定义：

Agent的Message就是指系统中“智能体”之间互相传递的指令或数据包

在 CAMEL 系统中，**BaseMessage 是所有消息对象的基础类，它为对话中的每一条信息提供了统一的结构和标准化的处理方式。**无论是用户输入的一段文本，还是包含图片、视频等多模态信息的数据包，都可以通过 BaseMessage 来统一表示和管理。

  1. 创建和使用Message

创建 BaseMessage 实例的最小化示例

from camel.messages import BaseMessage
from camel.types import RoleType

# 创建一个简单的用户消息
message = BaseMessage(
    role_name="test_user",
    role_type=RoleType.USER, #表示该消息来自用户，还有RoleType.ASSISTANT：表示该消息来自智能助手
    content="你好智能体!",#消息的核心载体，一般是文本，也可能是解析指令
    meta_dict={} #添加必需的meta dict参数，即使为空也要提供，否则会报 TypeError
)

print(message)

创建信息如下：

在这里插入图片描述

此时我们可以看到消息中还有类似于video_bytes, image_list 等 属性，由文生义，我们是可以附加其他类似视频、图像等多模态信息的

接下来我们尝试向信息中添加一张图片

# 下载一张图片并创建一个 PIL Image 对象
url = "https://raw.githubusercontent.com/camel-ai/camel/master/misc/logo_light.png"
response = requests.get(url)
img = Image.open(BytesIO(response.content))

# 创建一个简单的用户消息
message = BaseMessage(
    role_name="test_user",
    role_type=RoleType.USER,
    content="你好智能体!",
    meta_dict={},
    image_list=[img]  # 将图片列表作为参数传入
)

print(message)

  1. 不同消息的处理

    通过 BaseMessage 的类方法，我们可以快捷创建出用户（User）、助手（Assistant）的消息：

    from camel.messages import BaseMessage

# 创建用户消息
user_msg = BaseMessage.make_user_message(
    role_name="User_1",
    content="Hi, what can you do?"
)

# 创建助手消息
assistant_msg = BaseMessage.make_assistant_message(
    role_name="Assistant_1",
    content="I can help you with various tasks."
)

print("User Message:", user_msg)
print("Assistant Message:", assistant_msg)

    可以发现我们无需再手动指定role_type,保证了代码的简洁性

    更新消息内容，基于原有消息创建新消息

    # 基于用户消息创建一个新消息，内容稍作修改
updated_user_msg = user_msg.create_new_instance("Hi, can you tell me more about CAMEL?")
print("Updated User Message:", updated_user_msg)

    需要查看消息内部结构时候，BaseMessage 的 to_dict() 方法可以直接将消息对象转化为字典结构

  2. 与ChatAgent协作

ChatAgent 是 CAMEL 系统中负责对话处理与智能回应的组件。当你将 BaseMessage 对象传递给 ChatAgent 时，ChatAgent 将根据系统和用户消息的内容，生成具有上下文感知的回复。

仅想与智能助手进行一段简单的对话，可以直接构造一个文本类型的用户消息，并使用 step() 方法进行响应

示例代码：

model = ModelFactory.create(
    model_platform=ModelPlatformType.OPENAI_COMPATIBLE_MODEL,
    model_type="Qwen/Qwen2.5-72B-Instruct",
    url='https://api-inference.modelscope.cn/v1/',
    api_key=api_key
)

# 创建系统消息，告诉ChatAgent自己的角色定位
system_msg = "You are a helpful assistant that responds to user queries."

# 实例化一个ChatAgent
chat_agent = ChatAgent(model=model, system_message=system_msg,output_language='zh')

# 构造用户消息
user_msg = "Hello! Can you tell me something about CAMEL AI?"

# 将用户消息传给ChatAgent，并获取回复
response = chat_agent.step(user_msg) 
print("Assistant Response:", response.msgs[0].content)

当然我们也可以使用 BaseMessage 传递更丰富的上下文和多模态信息给 ChatAgent

示例：发送图片并获取回复

load_dotenv()
api_key = os.getenv('API_KEY')

model = ModelFactory.create(
    model_platform=ModelPlatformType.OPENAI_COMPATIBLE_MODEL,
    model_type="Qwen/QVQ-72B-Preview",
    url='https://api-inference.modelscope.cn/v1/',
    api_key=api_key
)

# 实例化ChatAgent
chat_agent = ChatAgent(model=model,output_language='中文')

# 下载一张图片并创建一个 PIL Image 对象
url = "https://img0.baidu.com/it/u=2205376118,3235587920&fm=253&fmt=auto&app=120&f=JPEG?w=846&h=800"
response = requests.get(url)
img = Image.open(BytesIO(response.content))

user_image_msg = BaseMessage.make_user_message(
    role_name="User",
    content="请描述这张图片的内容",
    image_list=[img]  # 将图片放入列表中
)

# 将包含图片的消息传给ChatAgent
response_with_image = chat_agent.step(user_image_msg)
print("Assistant's description of the image:", response_with_image.msgs[0].content)

回复如下：

在这里插入图片描述

  1. Agent的响应

    camel.responses 模块是 CAMEL 框架中处理聊天Agent响应的重要部分。其中ChatAgentResponse 类用于封装聊天Agent（ChatAgent）的交互输出，结构化响应内容，便于开发者访问消息、会话状态等信息。

ChatAgentResponse 的类属性包括：

  • msgs：一个包含 BaseMessage 对象的列表，表示Agent生成的消息。

  • terminated：一个布尔值，指示聊天会话是否已经被Agent终止。

  • info：一个字典，包含与会话相关的附加信息，例如使用统计或工具调用信息。

以下是一个ChatAgentResponse 实例

response = ChatAgentResponse(
    msgs=[
        BaseMessage(
            role_name="Assistant",  # 助手的角色名称
            role_type=RoleType.ASSISTANT,  # 指定角色类型
            content="你好，我可以帮您做什么？",  # 消息内容
            meta_dict={}  # 提供一个空的元数据字典（可根据需要填充）
        )
    ],  
    terminated=False,  # 会话未终止
    info={"usage": {"prompt_tokens": 10, "completion_tokens": 15}}  # 附加信息
)
五、提示词工程

可以参考openai的提示词建议：https://platform.openai.com/docs/guides/prompt-engineering/six-strategies-for-getting-better-results

通过精确的语言引导智能Agent提供更精准的服务和响应。

CoT（思维链）：让大模型逐步将一个复杂的问题分解为更简单的几个子问题，一步一步解决，直到得出答案。这个分解过程就被称为“思维链”

CoT的类型：

  • Zero-Shot-CoT：没有提供具体的例子，只是在问题中加一句“让我们一步步思考”
  • Few-Shot-CoT：给 AI 一些例子，每个例子都包含问题、推理过程和答案，像是给 AI 讲解解题步骤。

  1. CAMEL中的prompt

    CAMEL 提供了一些便捷的工具来帮助用户使用 CoT。我们可以使用 TaskSpecifyAgent 创建一个特定任务Agent，它会自动调用带有 CoT 的模板。

示例：

from camel.agents import TaskSpecifyAgent
from camel.models import ModelFactory
from camel.types import ModelPlatformType, TaskType
import os
from dotenv import load_dotenv

load_dotenv()
api_key = os.getenv('API_KEY')

model = ModelFactory.create(
    model_platform=ModelPlatformType.OPENAI_COMPATIBLE_MODEL,
    model_type="Qwen/Qwen2.5-72B-Instruct",
    url='https://api-inference.modelscope.cn/v1/',
    api_key=api_key
)
task_specify_agent = TaskSpecifyAgent(
    model=model, task_type=TaskType.AI_SOCIETY,output_language='中文'
)
specified_task_prompt = task_specify_agent.run(
    task_prompt="Improving stage presence and performance skills",
    meta_dict=dict(
        assistant_role="Musician", user_role="Student", word_limit=100
    ),
)
print(f"Specified task prompt:\n{specified_task_prompt}\n")

上述代码是通过设置角色和输出语言等元信息，基于原来的prompt生成更合适的prompt

在这里插入图片描述

除此之外，CAMEL 还允许用户创建自己的 Prompt 模板，使得生成的 Prompt 更加符合用户的需求。

例如：

my_prompt_template = TextPrompt(
    'Here is a task: I\'m a {occupation} and I want to {task}. Help me to make this task more specific.'
)  # 你可以根据需求自定义任何模板
task_specify_agent = TaskSpecifyAgent(
    model=model, task_specify_prompt=my_prompt_template
)
response = task_specify_agent.run(
    task_prompt="get promotion",
    meta_dict=dict(occupation="Software Engineer"),
)

注意：TextPrompt 类表示一个文本提示，并扩展了 str 类的功能。它提供了一个名为 key_words 的属性，表示提示中的关键词，比如上面代码的关键字是occupation和task。

format 方法允用提供的值替换格式字符串中的关键词。

from camel.prompts import TextPrompt
prompt = TextPrompt('Your name and age are: {name}, {age}')
name, age = 'John', 30
formatted_prompt = prompt.format(name=name, age=age)
print(formatted_prompt)

我们使用 format 方法将关键词 {name} 和 {age} 替换为值 ‘John’ 和 30。

六、Memory

Memory模块是一个关键的组件，其主要功能是存储和检索信息，以支持agent的学习和决策过程,能够动态地保存和更新信息，使agent能够利用过去的经验进行推理和决策

核心功能：

  • 信息存储
  • 信息检索
  • 记忆更新
  • 记忆管理

  1. ChatHistoryBlock

    ChatHistoryBlock 是一个基于键值存储的聊天历史记忆块实现。

参数说明

  • storage: 存储后端,默认使用InMemoryKeyValueStorage
  • keep_rate: 历史消息权重衰减率,默认 0.9
    • 该模块主要实现了以下方法：
  • retrieve()：使用可选的窗口大小获取最近的聊天记录
  • write_records()：将新记录写入聊天记录
  • clear()：删除所有聊天消息

keep_rate它主要用于调整历史消息在上下文中的重要性。

  1. 最新消息的 score 初始值为 1.0
  2. 每往前一条消息,score 会乘以 keep_rate
  3. system消息始终保持 score=1.0

示例用法

from camel.memories.blocks import ChatHistoryBlock
from camel.memories.records import MemoryRecord
from camel.types import OpenAIBackendRole
from camel.messages import BaseMessage

# 创建一个 ChatHistoryBlock 实例
chat_history = ChatHistoryBlock(keep_rate=0.8)

# 模拟写入一些消息记录
chat_history.write_records([
    MemoryRecord(message=BaseMessage.make_assistant_message(role_name="user", content="Hello,今天感觉怎么样？"), role_at_backend=OpenAIBackendRole.USER),
    MemoryRecord(message=BaseMessage.make_user_message(role_name="assistant", content="我很好，谢谢！"), role_at_backend=OpenAIBackendRole.ASSISTANT),
    MemoryRecord(message=BaseMessage.make_user_message(role_name="user", content="你能做些什么？"), role_at_backend=OpenAIBackendRole.USER),
    MemoryRecord(message=BaseMessage.make_assistant_message(role_name="assistant", content="我可以帮助你完成各种任务。"), role_at_backend=OpenAIBackendRole.ASSISTANT),
])

# 检索最近的 3 条消息
recent_records = chat_history.retrieve(window_size=4)

for record in recent_records:
    print(f"消息: {record.memory_record.message.content}, 权重: {record.score}")

  1. VectorDBBlock

    VectorDBBlock 是一个基于向量数据库的语义记忆块实现。

功能：

  • 使用向量存储后端

  • 支持语义相似度检索

  • 实现消息的向量化存储

  • retrieve()：根据关键字获取相似记录

  • write_records()：将新记录转换并写入矢量数据库

  • clear()：从向量数据库中删除所有记录

工作流程：

  1. 存储过程:

    • 将消息内容转换为向量表示

    • 生成唯一标识符（UUID）

    • 将向量和原始消息存入向量数据库

  2. 检索过程:

    • 将查询关键词转换为向量

    • 在向量空间中搜索相似向量

    • 返回相似度最高的记录

示例用法

from camel.memories.blocks.vectordb_block import VectorDBBlock
from camel.memories.records import MemoryRecord
from camel.messages import BaseMessage
from camel.embeddings import SentenceTransformerEncoder
from camel.types import OpenAIBackendRole

# 创建一个 VectorDBBlock 实例
vector_db_block = VectorDBBlock(embedding=SentenceTransformerEncoder(model_name="BAAI/bge-m3"))

# 创建一些示例聊天记录
records = [
    MemoryRecord(message=BaseMessage.make_user_message(role_name="user", content="今天天气真好！"), role_at_backend=OpenAIBackendRole.USER),
    MemoryRecord(message=BaseMessage.make_user_message(role_name="user", content="你喜欢什么运动？"), role_at_backend=OpenAIBackendRole.USER),
    MemoryRecord(message=BaseMessage.make_user_message(role_name="user", content="今天天气不错，我们去散步吧。"), role_at_backend=OpenAIBackendRole.USER),
]

# 将记录写入向量数据库
vector_db_block.write_records(records)

# 使用关键词进行检索
keyword = "天气"
retrieved_records = vector_db_block.retrieve(keyword=keyword, limit=3)

for record in retrieved_records:
    print(f"UUID: {record.memory_record.uuid}, Message: {record.memory_record.message.content}, Score: {record.score}")

运行结果如下：

在这里插入图片描述

我们可以看到相似度最高的前三条信息的得分

  1. CAMEL中的具体实践

CAMEL中目前支持key_value，graph，vector三种形式对于LLM信息进行存储，以供需要的时候检索。

下面是具体实践的示例

  • 创建memory对象，并写入对话记忆（告知CAMEL是第一个LLM多智能体框架）
from camel.memories import (
    LongtermAgentMemory,
    MemoryRecord,
    ScoreBasedContextCreator,
    ChatHistoryBlock,
    VectorDBBlock,
)
from camel.messages import BaseMessage
from camel.types import ModelType, OpenAIBackendRole
from camel.utils import OpenAITokenCounter
from camel.embeddings import SentenceTransformerEncoder

# 1. 初始化内存系统
memory = LongtermAgentMemory(
    context_creator=ScoreBasedContextCreator(
        token_counter=OpenAITokenCounter(ModelType.GPT_4O_MINI),
        token_limit=1024,
    ),
    chat_history_block=ChatHistoryBlock(),
    vector_db_block=VectorDBBlock(embedding=SentenceTransformerEncoder(model_name="BAAI/bge-m3")),
)

# 2. 创建记忆记录
records = [
    MemoryRecord(
        message=BaseMessage.make_user_message(
            role_name="User",
            content="什么是CAMEL AI?"
        ),
        role_at_backend=OpenAIBackendRole.USER,
    ),
    MemoryRecord(
        message=BaseMessage.make_assistant_message(
            role_name="Agent",
            content="CAMEL-AI是第一个LLM多智能体框架,并且是一个致力于寻找智能体 scaling law 的开源社区。"
        ),
        role_at_backend=OpenAIBackendRole.ASSISTANT,
    ),
]
# 3. 写入记忆
memory.write_records(records)

  • 创建agent，先让我们试试不导入记忆，直接询问，谁是第一个多智能体框架，看看回答如何

    #创建智能体
load_dotenv()
sys_msg = "你是一个好奇的智能体，正在探索宇宙的奥秘。"

api_key = os.getenv("API_KEY")

model = ModelFactory.create(
    model_platform=ModelPlatformType.OPENAI_COMPATIBLE_MODEL,
    model_type="Qwen/Qwen2.5-72B-Instruct",
    url='https://api-inference.modelscope.cn/v1/',
    api_key=api_key
)
agent = ChatAgent(model=model, sys_msg=sys_msg)
usr_msg = "告诉我基于我们讨论的内容，哪个是第一个LLM多智能体框架？"

response = agent.step(usr_msg)

print(response.msgs[0].content)

运行回答如下，可以看到并没有根据我们输入的记忆来回答：

在这里插入图片描述

  • 现在我们将memory对象赋值给agent的memory属性。
# 将memory赋值给agent
agent.memory = memory
# 发送消息给agent
response = agent.step(usr_msg)
# 查看响应
print(response.msgs[0].content)

我们再次运行 查看回答，可以发现这次它有根据我们输入的历史聊天记录进行回答：

在这里插入图片描述

七、TOOLS 工具

工具(Tools)是大语言模型(LLM)与外部世界交互的桥梁。配备工具可以避免产生幻觉

示例：定义数学工具

首先，定义加法函数并使用FunctionTool

from camel.toolkits import FunctionTool

def add(a, b) :
    return a + b

# 用 FunctionTool 包装该函数
add_tool = FunctionTool(add)

之后我们定义一下agent,并给他加上工具参数

model = ModelFactory.create(
    model_platform=ModelPlatformType.OPENAI_COMPATIBLE_MODEL,
    model_type="Qwen/Qwen2.5-72B-Instruct",
    url='https://api-inference.modelscope.cn/v1/',
    api_key=api_key
)
# 定义系统消息
sys_msg = "你是一个数学大师，擅长各种数学问题。当你遇到数学问题的时候，你要调用工具，将工具计算的结果作为答案"
usr_msg = "19987+2133等于多少？"
tool_agent = ChatAgent(
    tools=[add_tool],
    system_message=sys_msg,
    model=model,
    output_language="中文")

# 重新发送消息给toolagent
response = tool_agent.step(usr_msg)
print(response.msgs[0].content)

并且通过以下的命令可以检查工具是否真的被调用

print(response.info['tool_calls'])

模型回答正确并且可以看到工具确实被调用了

在这里插入图片描述

示例：进阶案例——小型的AI-Society系统

给Assistant Agent 配备相应的工具，duckduckgo搜索工具以消除幻觉

from camel.toolkits import SearchToolkit, MathToolkit
from camel.models import ModelFactory
from camel.types import ModelPlatformType
from camel.societies import RolePlaying
from camel.agents.chat_agent import FunctionCallingRecord
from camel.utils import print_text_animated
from colorama import Fore

import os
from dotenv import load_dotenv

load_dotenv()
api_key = os.getenv('API_KEY')

# 定义工具包
tools_list = [
    *SearchToolkit().get_tools(),
    *MathToolkit().get_tools()
]


#如果没有谷歌搜索API，可以使用duckduckgo工具，无需设置api即可使用
task_prompt = ("假设现在是2024年，"
        "牛津大学的成立年份，并计算出其当前年龄。"
        "然后再将这个年龄加上10年。使用duckduckgo搜索工具。")

# 创建模型
model = ModelFactory.create(
    model_platform=ModelPlatformType.OPENAI_COMPATIBLE_MODEL,
    model_type="Qwen/Qwen2.5-72B-Instruct",
    url='https://api-inference.modelscope.cn/v1/',
    api_key=api_key
)

# 设置角色扮演
role_play_session = RolePlaying(
    assistant_role_name="搜索者",
    user_role_name="教授",
    assistant_agent_kwargs=dict(
        model=model,
        tools=tools_list,
    ),
    user_agent_kwargs=dict(
        model=model,
    ),
    task_prompt=task_prompt,
    with_task_specify=False,
    output_language='中文'
)

# 设置聊天轮次限制
chat_turn_limit = 10

print(
    Fore.GREEN
    + f"AI助手系统消息:\n{role_play_session.assistant_sys_msg}\n"
)
print(
    Fore.BLUE + f"AI用户系统消息:\n{role_play_session.user_sys_msg}\n"
)

print(Fore.YELLOW + f"原始任务提示:\n{task_prompt}\n")
print(
    Fore.CYAN
    + "指定的任务提示:"
    + f"\n{role_play_session.specified_task_prompt}\n"
)
print(Fore.RED + f"最终任务提示:\n{role_play_session.task_prompt}\n")

n = 0
input_msg = role_play_session.init_chat()
while n < chat_turn_limit:
    n += 1
    assistant_response, user_response = role_play_session.step(input_msg)

    if assistant_response.terminated:
        print(
            Fore.GREEN
            + (
                "AI助手终止。原因: "
                f"{assistant_response.info['termination_reasons']}."
            )
        )
        break
    if user_response.terminated:
        print(
            Fore.GREEN
            + (
                "AI用户终止。"
                f"原因: {user_response.info['termination_reasons']}."
            )
        )
        break

    # 打印用户的输出
    print_text_animated(
        Fore.BLUE + f"AI用户:\n\n{user_response.msg.content}\n"
    )

    if "CAMEL_TASK_DONE" in user_response.msg.content:
        break

    # 打印助手的输出，包括任何函数执行信息
    print_text_animated(Fore.GREEN + "AI助手:")
    tool_calls: list[FunctionCallingRecord] = assistant_response.info[
        'tool_calls'
    ]
    for func_record in tool_calls:
        print_text_animated(f"{func_record}")
    print_text_animated(f"{assistant_response.msg.content}\n")

    input_msg = assistant_response.msg

运行效果如下：

在这里插入图片描述

在这里插入图片描述

我们可以看到通过两个AI 角色的对话和搜索工具逐步完成了任务

# python # 人工智能
Logo
AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

cover

最强开源通用智能体Suna部署调用实战—100%吊打Manus

avatar AI Agent技术社区
cover

【2025年5月】AI生产力再探再报：各家智能体持续内卷，前沿应用不断细分

avatar AI Agent技术社区
cover

AIGC学习笔记（8）——AI大模型开发工程师

avatar AI Agent技术社区