一步步教你使用 Agent Zero：AI 智能代理开发实战完全指南

从零开始构建你的第一个 AI Agent 应用

前言

在人工智能快速发展的今天，AI Agent（人工智能代理）已经成为开发者社区中最热门的话题之一。Agent Zero 是一个开源的 AI Agent 框架，它以其灵活性和强大的功能吸引了超过 17,000 颗星标。本文将带领你从环境搭建开始，一步步掌握 Agent Zero 的核心功能，并通过实际案例让你能够独立开发基于 Agent Zero 的智能应用。无论你是刚刚接触 AI 开发的新手，还是希望深入了解 Agent 技术的资深开发者，这篇教程都将为你提供全面而实用的指导。

第一章：环境搭建与基础配置

在开始使用 Agent Zero 之前，我们需要先完成开发环境的准备工作。一个正确的环境配置是后续学习和开发的基础，下面我们将详细介绍从安装到配置的完整流程。

1.1 系统要求与依赖准备

Agent Zero 是一个基于 Python 的项目，因此对系统环境有一定的要求。首先，你需要确保你的电脑上安装了 Python 3.10 或更高版本。你可以通过在终端或命令提示符中输入以下命令来检查当前的 Python 版本：

python --version

如果显示的版本低于 3.10，建议你前往 Python 官方网站下载并安装最新版本的 Python。安装完成后，我们还需要确保 pip（Python 的包管理工具）已经正确安装并更新到最新版本。你可以使用以下命令来升级 pip：

python -m pip install --upgrade pip

对于不同的操作系统，具体的安装步骤可能会有所不同。在 Windows 系统上，你需要确保 Python 已经添加到系统的环境变量中。在 macOS 或 Linux 系统上，你可以使用包管理器（如 Homebrew 或 apt）来安装 Python。无论使用哪种操作系统，强烈建议你使用虚拟环境来隔离项目依赖，这样可以避免不同项目之间的依赖冲突。

1.2 创建虚拟环境

虚拟环境是 Python 开发中的最佳实践，它可以让你为每个项目创建独立的依赖环境。下面我们来创建并激活一个专门用于 Agent Zero 项目的虚拟环境。首先，打开你的终端并进入你想要存放项目的目录，然后执行以下命令创建虚拟环境：

python -m venv agent-zero-env

创建完成后，在 Windows 系统上你需要通过以下命令来激活虚拟环境：

agent-zero-env\Scripts\activate

在 macOS 或 Linux 系统上，激活命令则是：

source agent-zero-env/bin/activate

激活成功后，你会在终端提示符前面看到虚拟环境的名称（agent-zero-env），这表示你已经成功进入了虚拟环境。在这个环境中安装的任何包都不会影响系统全局的 Python 环境或其他项目。

1.3 安装 Agent Zero

现在虚拟环境已经准备就绪，接下来我们开始安装 Agent Zero。Agent Zero 的安装非常简单，你可以通过 pip 直接从 PyPI 安装，也可以从 GitHub 仓库克隆最新版本进行安装。

使用 pip 直接安装是最简单快捷的方式，只需要执行以下命令：

pip install agent-zero

如果你想要安装最新开发版本，或者希望对代码进行自定义修改，可以选择从 GitHub 克隆仓库。首先，确保你的电脑上已经安装了 Git。然后执行以下命令克隆仓库：

git clone https://github.com/agent0ai/agent-zero.git

克隆完成后，进入项目目录并使用 pip 安装：

cd agent-zero
pip install -e .

使用 pip install -e . 命令进行可编辑模式安装的好处是，你对源代码的任何修改都会立即生效，无需重新安装。这对于想要深入研究框架源码或进行二次开发的用户来说非常方便。

1.4 配置 API 密钥

Agent Zero 需要连接外部的 AI 模型服务来完成各种任务。你需要配置相应的 API 密钥才能让 Agent 正常工作。目前 Agent Zero 支持多种 AI 服务提供商，包括 OpenAI、Anthropic、Google 等。

最简单的方式是通过环境变量来配置 API 密钥。你可以在终端中直接设置：

export OPENAI_API_KEY="你的OpenAI API密钥"
export ANTHROPIC_API_KEY="你的Anthropic API密钥"

或者，如果你希望这些设置永久生效，可以将它们添加到虚拟环境或系统级别的配置文件中。在虚拟环境中，你可以创建一个 .env 文件来存储这些敏感信息：

# 在项目根目录创建 .env 文件
# .env 文件内容示例

# OpenAI API 配置
OPENAI_API_KEY=sk-xxxxxxxxxxxxxxxxxxxxxxxx

# Anthropic API 配置（可选）
ANTHROPIC_API_KEY=sk-ant-xxxxxxxxxxxxxxxxxxxxxxxx

# 默认使用的模型
DEFAULT_MODEL=gpt-4

# 日志级别设置
LOG_LEVEL=INFO

需要注意的是，.env 文件包含了敏感信息，不应该将其提交到版本控制系统（如 Git）中。确保在项目的 .gitignore 文件中添加 .env 以防止意外泄露。

1.5 验证安装

安装完成后，我们需要验证 Agent Zero 是否已经正确安装并能够正常工作。你可以创建一个简单的测试脚本来验证：

# test_installation.py
# 这是我们用于验证 Agent Zero 安装的测试脚本

print("开始验证 Agent Zero 安装...")

try:
    # 尝试导入 agent_zero 模块
    import agent_zero

    # 打印版本信息
    print(f"Agent Zero 版本: {agent_zero.__version__}")
    print("模块导入成功！")

except ImportError as e:
    print(f"导入失败: {e}")
    print("请检查 Agent Zero 是否正确安装")

except Exception as e:
    print(f"验证过程中出现错误: {e}")

运行这个测试脚本：

python test_installation.py

如果看到成功导入的消息，恭喜你，Agent Zero 已经正确安装并准备就绪。

第二章：核心概念与架构解析

在开始实际使用 Agent Zero 之前，我们需要先理解它的核心概念和整体架构。了解这些基础知识将帮助你更好地使用框架的各项功能，并能够根据实际需求进行定制和扩展。

2.1 Agent Zero 的设计理念

Agent Zero 是一个基于大型语言模型（LLM）的智能代理框架。与传统的规则驱动型代理不同，Agent Zero 利用 LLM 的自然语言理解和生成能力，让代理能够理解人类指令、进行推理、规划行动，并执行各种复杂任务。

Agent Zero 的设计理念可以概括为以下几点：

第一，简洁至上。框架尽可能减少用户的配置负担，通过合理的默认设置让用户能够快速上手。同时，框架也提供了丰富的配置选项，让高级用户能够进行深度定制。

第二，工具即代码。在 Agent Zero 中，代理可以使用的一切工具都是以 Python 函数的形式定义的。这种设计让开发者可以使用熟悉的编程方式来扩展代理的能力，无需学习新的配置语法或领域特定语言。

第三，可观测性。框架内置了详细的日志和追踪功能，让用户能够清楚地了解代理的思考过程和执行步骤。这对于调试和优化代理行为非常重要。

第四，安全隔离。Agent Zero 提供了沙盒执行环境，确保代理的操作不会对主机系统造成损害。用户可以精确控制代理可以访问的资源和可以执行的操作。

2.2 核心组件详解

Agent Zero 的架构由几个核心组件构成，每个组件都有其特定的职责和功能。

Agent（代理） 是整个框架的核心概念。代理是一个能够自主决策、执行任务的智能体。在 Agent Zero 中，每个代理都有自己的名称、角色描述、可用工具列表和记忆系统。代理接收用户输入，分析任务，制定计划，然后调用工具执行任务。

# agent_example.py
# 演示如何创建一个基本的 Agent 实例

from agent_zero import Agent

# 创建第一个代理实例
my_agent = Agent(
    name="写作助手",
    role="一位专业的写作助手，帮助用户撰写各种类型的内容",
    tools=["search", "write_document"],
    model="gpt-4"
)

# 代理创建完成，准备接受任务
print("代理创建成功！")

Tools（工具） 是代理与外部世界交互的桥梁。工具可以是任何可调用的函数，从简单的计算器到复杂的 API 调用都可以作为工具使用。代理能够根据任务需求自主选择和调用合适的工具。

Memory（记忆） 系统让代理能够记住之前的对话和操作。Agent Zero 提供了多种记忆实现方式，包括短期记忆（当前会话）和长期记忆（持久化存储）。通过记忆系统，代理能够保持上下文的连贯性，提供更加个性化的服务。

Executor（执行器） 负责运行代理执行的任务。Agent Zero 提供了多种执行器实现，支持同步和异步执行模式，以及不同的执行策略（如顺序执行、并行执行等）。

2.3 工作流程解析

理解 Agent Zero 的工作流程对于有效使用框架至关重要。整个工作流程可以分为以下几个阶段：

第一阶段：输入解析。当用户发送一条指令时，代理首先解析这条指令，理解用户想要完成的任务是什么。这个阶段会进行意图识别和关键信息提取。

第二阶段：任务规划。基于对任务的理解，代理会制定执行计划。这个计划可能包括需要调用的工具、执行顺序、需要的参数等。在复杂任务中，代理可能会将大任务分解为多个子任务。

第三阶段：工具执行。代理按照计划依次执行各个步骤。每一步执行后，代理会评估结果，判断是否需要调整后续计划。这个过程是迭代进行的，直到任务完成。

第四阶段：结果汇总。任务完成后，代理将执行结果整理成用户友好的格式进行返回。同时，相关的信息会被存储到记忆系统中，以备后续使用。

# workflow_visual.py
# 展示 Agent Zero 工作流程的伪代码

class AgentWorkflow:
    """
    Agent Zero 工作流程演示
    展示了从接收指令到返回结果的完整过程
    """

    def process_user_input(self, user_input):
        # 阶段一：输入解析
        parsed_input = self.parse_input(user_input)

        # 阶段二：任务规划
        plan = self.create_plan(parsed_input)

        # 阶段三：工具执行
        results = self.execute_plan(plan)

        # 阶段四：结果汇总
        final_response = self.format_response(results)

        return final_response

2.4 消息与对话系统

Agent Zero 使用消息系统来管理代理与用户之间的交互。每条消息都有类型、发送者和内容。常见的消息类型包括：

UserMessage（用户消息） 表示来自最终用户的消息，这是代理接收输入的主要方式。

AssistantMessage（助手消息） 表示代理生成的回复，包含文本内容和可能的工具调用请求。

ToolMessage（工具消息） 表示工具执行的结果，当代理调用工具时，会生成工具消息来传递执行结果。

SystemMessage（系统消息） 用于传递系统级指令或配置，如系统提示、角色设定等。

# message_system.py
# 演示 Agent Zero 中的消息系统

from agent_zero import Agent, UserMessage, SystemMessage

# 创建带有系统提示的代理
my_agent = Agent(
    name="客服助手",
    role="一位友好专业的在线客服",
    system_prompt="你是一家电商平台的客服助手，"
                  "需要热情、耐心地回答顾客的问题。"
)

# 创建用户消息
user_message = UserMessage(
    content="我想查询一下我的订单什么时候能到？订单号是 12345678"
)

# 发送消息给代理
response = my_agent.send_message(user_message)

# 打印代理的回复
print(f"代理回复: {response.content}")

第三章：实战教程 —— 第一个 Agent Zero 项目

现在我们已经了解了 Agent Zero 的基本概念和架构，接下来让我们通过一个完整的实战项目来学习如何使用这个框架。我们将创建一个多功能助手代理，它能够帮助你完成日常工作中的各种任务。

3.1 项目规划与设计

在开始编码之前，我们先来规划这个项目。我们希望创建一个能够帮助用户完成以下任务的助手：

第一，回答用户的一般性问题（使用 LLM 的知识）。

第二，帮助用户搜索网络信息。

第三，帮助用户管理待办事项。

第四，记录和分析用户的工作日志。

基于这些需求，我们需要定义相应的工具，并设计代理的角色和对话流程。

3.2 定义自定义工具

工具是 Agent Zero 中最重要的扩展机制。一个工具就是一个 Python 函数，框架会自动将其转换为代理可以调用的能力。下面让我们定义几个实用的工具：

# tools.py
# 定义我们的自定义工具集合

import json
from datetime import datetime, timedelta
from typing import List, Dict, Any

# ============================================================
# 工具一：待办事项管理
# ============================================================

class TodoManager:
    """
    待办事项管理器
    提供添加、查看、完成和删除待办事项的功能
    """

    def __init__(self):
        # 使用字典存储待办事项，键为ID，值为事项详情
        self.todos = {}
        self.next_id = 1

    def add_todo(self, task: str, priority: str = "medium", 
                 due_date: str = None) -> str:
        """
        添加新的待办事项

        参数:
            task: 任务描述
            priority: 优先级，可选 low/medium/high
            due_date: 截止日期，格式为 YYYY-MM-DD

        返回:
            新创建事项的ID
        """
        todo_id = str(self.next_id)
        self.next_id += 1

        todo_item = {
            "id": todo_id,
            "task": task,
            "priority": priority,
            "due_date": due_date,
            "status": "pending",
            "created_at": datetime.now().isoformat()
        }

        self.todos[todo_id] = todo_item

        return json.dumps({
            "success": True,
            "message": f"已添加待办事项 [{todo_id}]: {task}",
            "todo_id": todo_id
        })

    def list_todos(self, status: str = None) -> str:
        """
        列出所有待办事项

        参数:
            status: 筛选状态，可选 pending/completed/all

        返回:
            待办事项列表的JSON字符串
        """
        if status == "all" or status is None:
            filtered_todos = self.todos.values()
        else:
            filtered_todos = [
                t for t in self.todos.values() 
                if t["status"] == status
            ]

        result = {
            "total": len(filtered_todos),
            "todos": sorted(
                filtered_todos, 
                key=lambda x: (x["priority"], x["created_at"]),
                reverse=True
            )
        }

        return json.dumps(result, ensure_ascii=False, indent=2)

    def complete_todo(self, todo_id: str) -> str:
        """
        将待办事项标记为完成

        参数:
            todo_id: 待办事项的ID
        """
        if todo_id not in self.todos:
            return json.dumps({
                "success": False,
                "message": f"未找到ID为 {todo_id} 的待办事项"
            })

        self.todos[todo_id]["status"] = "completed"
        self.todos[todo_id]["completed_at"] = datetime.now().isoformat()

        return json.dumps({
            "success": True,
            "message": f"已完成: {self.todos[todo_id]['task']}"
        })

    def delete_todo(self, todo_id: str) -> str:
        """
        删除待办事项

        参数:
            todo_id: 待办事项的ID
        """
        if todo_id not in self.todos:
            return json.dumps({
                "success": False,
                "message": f"未找到ID为 {todo_id} 的待办事项"
            })

        deleted_task = self.todos[todo_id]["task"]
        del self.todos[todo_id]

        return json.dumps({
            "success": True,
            "message": f"已删除: {deleted_task}"
        })

# 创建全局的待办事项管理器实例
todo_manager = TodoManager()

# ============================================================
# 工具二：简单计算器
# ============================================================

def calculator(expression: str) -> str:
    """
    执行基本的数学计算

    参数:
        expression: 数学表达式，如 "2 + 3 * 4"

    返回:
        计算结果
    """
    try:
        # 安全评估：只允许基本数学运算
        # 注意：实际使用中应该使用更安全的表达式解析器
        allowed_chars = set("0123456789+-*/.() ")
        if not all(c in allowed_chars for c in expression):
            return json.dumps({
                "success": False,
                "error": "表达式包含非法字符"
            })

        result = eval(expression)

        return json.dumps({
            "success": True,
            "expression": expression,
            "result": result
        })

    except Exception as e:
        return json.dumps({
            "success": False,
            "error": f"计算错误: {str(e)}"
        })

# ============================================================
# 工具三：文本处理工具
# ============================================================

def text_processor(text: str, operation: str) -> str:
    """
    对文本进行处理操作

    参数:
        text: 输入文本
        operation: 操作类型，可选 word_count/char_count/
                   uppercase/lowercase/reverse/summary
    """
    operations = {
        "word_count": lambda t: len(t.split()),
        "char_count": lambda t: len(t),
        "uppercase": lambda t: t.upper(),
        "lowercase": lambda t: t.lower(),
        "reverse": lambda t: t[::-1]
    }

    if operation == "summary":
        # 返回文本摘要信息
        return json.dumps({
            "success": True,
            "word_count": len(text.split()),
            "char_count": len(text),
            "char_count_no_space": len(text.replace(" ", "")),
            "lines": text.count("\n") + 1
        })

    if operation not in operations:
        return json.dumps({
            "success": False,
            "error": f"不支持的操作: {operation}"
        })

    result = operations[operation](text)

    return json.dumps({
        "success": True,
        "operation": operation,
        "input_length": len(text),
        "result": result
    })

# ============================================================
# 工具四：日期时间工具
# ============================================================

def date_time_handler(operation: str, date_str: str = None) -> str:
    """
    处理日期时间相关的操作

    参数:
        operation: 操作类型，可选 now/days_until/add_days/diff
        date_str: 日期字符串，格式为 YYYY-MM-DD
    """
    from datetime import datetime

    if operation == "now":
        now = datetime.now()
        return json.dumps({
            "success": True,
            "current_datetime": now.isoformat(),
            "formatted": now.strftime("%Y年%m月%d日 %H:%M:%S")
        })

    if date_str:
        try:
            target_date = datetime.strptime(date_str, "%Y-%m-%d")
        except ValueError:
            return json.dumps({
                "success": False,
                "error": "日期格式错误，请使用 YYYY-MM-DD 格式"
            })
    else:
        target_date = datetime.now()

    if operation == "days_until":
        now = datetime.now()
        diff = (target_date - now).days
        return json.dumps({
            "success": True,
            "target_date": date_str,
            "days_until": diff,
            "message": f"距离 {date_str} 还有 {diff} 天" if diff >= 0 else f"{date_str} 是 {-diff} 天前"
        })

    if operation == "add_days":
        days = int(date_str) if isinstance(date_str, int) else 7
        new_date = target_date + timedelta(days=days)
        return json.dumps({
            "success": True,
            "original_date": target_date.isoformat(),
            "days_added": days,
            "new_date": new_date.strftime("%Y-%m-%d")
        })

    return json.dumps({
        "success": False,
        "error": f"不支持的操作: {operation}"
    })

3.3 创建主程序

现在我们已经定义好了工具，接下来创建主程序，将所有组件整合在一起：

# main.py
# Agent Zero 助手主程序

from agent_zero import Agent
from tools import (
    todo_manager,
    calculator,
    text_processor,
    date_time_handler
)

# ============================================================
# 初始化 Agent
# ============================================================

# 定义代理的系统提示
SYSTEM_PROMPT = """
你是一位全能助手，名为"小帮手"。你的职责是：
1. 热情友好地与用户交流
2. 准确理解用户的需求
3. 使用你拥有的工具帮助用户完成任务
4. 在回复中清晰地展示结果

你的可用工具包括：
- 待办事项管理：添加、查看、完成、删除待办事项
- 计算器：执行数学计算
- 文本处理：字数统计、格式转换等
- 日期时间：查询当前时间、日期计算等

当用户提出请求时，先分析需求，选择合适的工具完成任务。
"""

# 创建 Agent 实例
assistant = Agent(
    name="小帮手",
    role="全能助手",
    system_prompt=SYSTEM_PROMPT,
    model="gpt-4",
    verbose=True  # 开启详细日志输出
)

# 注册工具
assistant.register_tool("add_todo", todo_manager.add_todo)
assistant.register_tool("list_todos", todo_manager.list_todos)
assistant.register_tool("complete_todo", todo_manager.complete_todo)
assistant.register_tool("delete_todo", todo_manager.delete_todo)
assistant.register_tool("calculator", calculator)
assistant.register_tool("text_processor", text_processor)
assistant.register_tool("date_time", date_time_handler)

print("=" * 60)
print("欢迎使用小帮手全能助手！")
print("=" * 60)
print("\n我可以帮助你完成以下任务：")
print("  - 管理待办事项")
print("  - 执行数学计算")
print("  - 处理文本")
print("  - 查询日期时间")
print("\n请输入你的问题或指令，输入 '退出' 结束对话。")
print("-" * 60)

# ============================================================
# 对话循环
# ============================================================

def run_conversation():
    """
    运行对话循环，处理用户输入
    """
    conversation_history = []

    while True:
        try:
            # 获取用户输入
            user_input = input("\n你: ").strip()

            # 检查退出条件
            if user_input.lower() in ["退出", "exit", "quit"]:
                print("\n感谢使用小帮手，再见！")
                break

            # 忽略空输入
            if not user_input:
                continue

            # 将用户消息添加到历史
            conversation_history.append({
                "role": "user",
                "content": user_input
            })

            # 发送消息给代理并获取回复
            print("\n思考中...")
            response = assistant.send_message(user_input)

            # 显示代理回复
            print(f"\n小帮手: {response.content}")

            # 如果有工具调用，显示工具结果
            if response.tool_calls:
                print("\n" + "-" * 40)
                print("工具调用详情:")
                for tool_call in response.tool_calls:
                    print(f"  - {tool_call.tool_name}: {tool_call.arguments}")
                print("-" * 40)

            # 添加助手回复到历史
            conversation_history.append({
                "role": "assistant",
                "content": response.content
            })

        except KeyboardInterrupt:
            print("\n\n程序被用户中断，再见！")
            break
        except Exception as e:
            print(f"\n发生错误: {e}")
            print("请重试或输入 '退出' 结束程序。")

# 程序入口
if __name__ == "__main__":
    run_conversation()

运行这个程序，你就拥有了一个可以与你对话的智能助手。你可以让它帮你添加待办事项、进行计算、处理文本等。

3.4 运行和测试

让我们来测试一下这个助手程序：

$ python main.py

============================================================
欢迎使用小帮手全能助手！
============================================================

我可以帮助你完成以下任务：
  - 管理待办事项
  - 执行数学计算
  - 处理文本
  - 查询日期时间

请输入你的问题或指令，输入 '退出' 结束对话。
------------------------------------------------------------

你: 帮我添加三个待办事项

思考中...

小帮手: 当然可以！让我为你添加三个待办事项：

1. 完成项目报告 - 高优先级
2. 购买生活用品 - 中优先级  
3. 阅读新书章节 - 低优先级

我正在为你添加这些事项...

已完成添加！

你: 我的待办事项有哪些？

思考中...

小帮手: 这是你的待办事项列表：

待办事项 #1
  任务：完成项目报告
  优先级：高
  状态：待处理
  创建时间：2024-01-15

待办事项 #2
  任务：购买生活用品
  优先级：中
  状态：待处理
  创建时间：2024-01-15

待办事项 #3
  任务：阅读新书章节
  优先级：低
  状态：待处理
  创建时间：2024-01-15

你: 1024 除以 32 再加 17 等于多少？

思考中...

小帮手: 让我来计算：

1024 ÷ 32 + 17 = 32 + 17 = 49

答案是 49！

你: 退出

感谢使用小帮手，再见！

第四章：高级功能与进阶用法

掌握了基础用法之后，让我们继续探索 Agent Zero 的高级功能。这些功能将帮助你构建更加复杂和强大的智能代理应用。

4.1 多代理协作系统

在实际应用中，单个代理可能无法完成所有任务。Agent Zero 支持多代理协作，让多个专业代理协同工作来解决复杂问题。

# multi_agent.py
# 演示多代理协作系统

from agent_zero import Agent, MultiAgentSystem

# ============================================================
# 定义专业代理
# ============================================================

# 写作代理 - 专门负责内容创作
writer_agent = Agent(
    name="写作助手",
    role="专业的内容创作者",
    system_prompt="你是一位资深的内容创作者，擅长撰写各类文章。"
                  "你的职责是根据用户需求创作高质量的文本内容。",
    tools=["text_generator", "grammar_check"]
)

# 研究代理 - 专门负责信息检索和分析
researcher_agent = Agent(
    name="研究助手",
    role="专业的研究分析师",
    system_prompt="你是一位专业的研究分析师，擅长信息检索、数据分析"
                  "和趋势预测。你的职责是为决策提供可靠的信息支持。",
    tools=["web_search", "data_analyzer"]
)

# 审核代理 - 专门负责内容审核和质量把控
reviewer_agent = Agent(
    name="审核助手",
    role="专业的质量审核员",
    system_prompt="你是一位严谨的质量审核员，负责检查内容的准确性、"
                  "完整性和专业性。",
    tools=["fact_checker", "plagiarism_detector"]
)

# ============================================================
# 创建多代理协作系统
# ============================================================

class ContentPipeline:
    """
    内容创作流水线
    演示如何让多个代理协同工作
    """

    def __init__(self):
        # 初始化各代理
        self.writer = writer_agent
        self.researcher = researcher_agent
        self.reviewer = reviewer_agent

        # 定义工作流程
        self.workflow = {
            "research": self.research_phase,
            "write": self.write_phase,
            "review": self.review_phase
        }

    def research_phase(self, topic):
        """研究阶段：收集相关信息"""
        print("阶段一：研究阶段")
        print(f"正在研究主题：{topic}")

        # 调用研究代理收集信息
        research_result = self.researcher.send_message(
            f"请帮我研究'{topic}'，提供关键信息和数据来源"
        )

        return research_result.content

    def write_phase(self, topic, research_data):
        """写作阶段：根据研究结果创作内容"""
        print("\n阶段二：写作阶段")
        print(f"正在根据研究结果创作关于'{topic}'的文章")

        # 调用写作代理创作文章
        article = self.writer.send_message(
            f"根据以下研究资料，撰写一篇关于'{topic}'的文章：\n\n{research_data}"
        )

        return article.content

    def review_phase(self, article):
        """审核阶段：检查内容质量"""
        print("\n阶段三：审核阶段")
        print("正在审核文章质量...")

        # 调用审核代理检查文章
        review_result = self.reviewer.send_message(
            f"请审核以下文章的质量和准确性：\n\n{article}"
        )

        return review_result.content

    def run_pipeline(self, topic):
        """
        运行完整的内容创作流水线

        参数:
            topic: 文章主题

        返回:
            最终发布的文章内容
        """
        # 执行各阶段
        research = self.research_phase(topic)
        article = self.write_phase(topic, research)
        review = self.review_phase(article)

        # 整合最终结果
        final_result = {
            "topic": topic,
            "research_summary": research,
            "article": article,
            "review_feedback": review
        }

        return final_result

# 使用示例
def demo_multi_agent():
    """演示多代理协作"""

    pipeline = ContentPipeline()

    # 运行创作流水线
    result = pipeline.run_pipeline("人工智能在医疗领域的应用")

    print("\n" + "=" * 60)
    print("内容创作完成！")
    print("=" * 60)
    print(f"\n主题: {result['topic']}")
    print(f"\n审核反馈: {result['review_feedback']}")

demo_multi_agent()

4.2 长期记忆系统

Agent Zero 提供了强大的记忆系统，支持短期记忆和长期记忆。长期记忆可以让代理跨越多个会话记住重要信息。

# memory_system.py
# 演示 Agent Zero 的记忆系统

from agent_zero import Agent
from agent_zero.memory import VectorMemory, FileMemory

# ============================================================
# 初始化带记忆功能的 Agent
# ============================================================

# 创建支持向量检索的记忆系统
vector_memory = VectorMemory(
    embedding_model="text-embedding-ada-002",
    vector_dim=1536
)

# 创建支持持久化存储的记忆系统
file_memory = FileMemory(
    storage_path="./memory_data",
    max_entries=1000
)

# 创建带有记忆功能的代理
memory_agent = Agent(
    name="记忆助手",
    role="私人助理",
    system_prompt="你是一个细心周到的私人助理，会记住用户的重要信息，"
                  "并在合适的时机使用这些信息提供个性化服务。",
    memory=vector_memory,
    verbose=True
)

# ============================================================
# 记忆操作函数
# ============================================================

def remember_user_info(agent, key, value):
    """
    记住用户信息

    参数:
        agent: Agent 实例
        key: 信息标签（如 "姓名"、"偏好" 等）
        value: 信息内容
    """
    memory_entry = {
        "type": "user_info",
        "key": key,
        "value": value,
        "timestamp": "current_time"
    }

    agent.remember(memory_entry)

    return f"已记住：{key} = {value}"

def recall_memories(agent, query, limit=5):
    """
    检索相关记忆

    参数:
        agent: Agent 实例
        query: 查询内容
        limit: 返回的记忆数量

    返回:
        相关记忆列表
    """
    related_memories = agent.recall(query, limit=limit)

    return related_memories

# ============================================================
# 演示：用户信息管理
# ============================================================

def demo_memory_system():
    """演示记忆系统的使用"""

    # 记住用户的基本信息
    print("训练阶段：让代理记住用户信息")
    print("-" * 40)

    remember_user_info(memory_agent, "姓名", "张三")
    remember_user_info(memory_agent, "职业", "软件工程师")
    remember_user_info(memory_agent, "兴趣", ["编程", "阅读", "音乐"])
    remember_user_info(memory_agent, "过敏原", "花生")

    print("已记住用户的基本信息\n")

    # 测试记忆检索
    print("检索阶段：测试记忆检索")
    print("-" * 40)

    # 当用户询问相关问题时，代理会自动使用记忆
    question = "我刚才说的我叫什么名字？"
    response = memory_agent.send_message(question)

    print(f"\n问题：{question}")
    print(f"回答：{response.content}")

    # 直接检索记忆
    print("\n直接检索记忆：")
    memories = recall_memories(memory_agent, "职业和兴趣")
    for mem in memories:
        print(f"  - {mem['key']}: {mem['value']}")

demo_memory_system()

4.3 异步执行与并发任务

对于需要处理大量任务的应用，Agent Zero 支持异步执行和并发处理：

# async_agent.py
# 演示异步执行功能

import asyncio
from agent_zero import AsyncAgent

# ============================================================
# 创建异步 Agent
# ============================================================

async_agent = AsyncAgent(
    name="异步助手",
    role="高效的异步任务处理器",
    system_prompt="你是一个高效的助手，可以同时处理多个任务。",
    tools=["web_search", "file_processor"]
)

# ============================================================
# 异步任务处理示例
# ============================================================

async def process_multiple_queries():
    """
    同时处理多个查询任务
    演示异步执行的优势
    """

    # 定义要同时处理的任务
    tasks = [
        "今天北京的天气怎么样？",
        "帮我总结一下人工智能的发展历史",
        "给我讲一个关于太空探索的笑话",
        "解释一下什么是量子计算",
        "推荐五本科很好的科幻小说"
    ]

    print("开始并发处理多个任务...")
    print(f"任务总数: {len(tasks)}")
    print("-" * 40)

    # 使用 asyncio 并发执行多个任务
    results = await async_agent.process_batch(tasks)

    # 打印结果
    for i, (task, result) in enumerate(zip(tasks, results), 1):
        print(f"\n任务 {i}: {task[:20]}...")
        print(f"结果: {result.content[:100]}...")

    return results

async def streaming_demo():
    """
    流式输出演示
    展示如何实时获取代理的思考过程
    """

    print("流式输出演示")
    print("-" * 40)

    async for chunk in async_agent.stream_message(
        "请详细介绍一下机器学习的基本概念"
    ):
        # 实时打印输出
        print(chunk, end="", flush=True)

    print("\n")

# 运行异步演示
async def main():
    await process_multiple_queries()
    await streaming_demo()

# 执行
# asyncio.run(main())

print("异步执行演示完成")

4.4 自定义执行策略

Agent Zero 允许你自定义代理的执行策略，以适应不同的应用场景：

“`python

custom_strategy.py

演示自定义执行策略

from agent_zero import Agent
from agent_zero.executor import (
SequentialExecutor,
ParallelExecutor,
RetryExecutor,
TimeoutExecutor
)

============================================================

定义执行策略

============================================================

class SafetyFirstExecutor(TimeoutExecutor):
“””
安全优先执行器
在执行每个操作前进行安全检查
“””

def __init__(self, max_retries=3, timeout=30):
    super().__init__(max_retries=max_retries, timeout=timeout)
    self.safety_checks = []

def add_safety_check(self, check_func):
    """添加安全检查函数"""
    self.safety_checks.append(check_func)

async def execute_tool(self, tool_call):
    """执行工具前进行安全检查"""

    # 执行所有安全检查
    for check in self.safety_checks:
        if not await check(tool_call):

Project: https://github.com/agent0ai/agent-zero

Stars: 17688