🚀 想用不到80行的Python代码构建一个无敌的AI代理吗？

本教程教你如何使用OpenAI的Swarm框架构建一个可靠的AI客服代理，然后通过DBOS持久性执行来增强其稳定性以防止崩溃，使其稳定可靠 💪🐝

这个代理会接收用户的名字，为用户处理退款，然后应用折扣。如果代理在处理退款的过程中被打断，重新启动后，它会自动恢复，完成退款，然后继续下一步的工作流程。

运行一下这个代理，并在任何时候按下Ctrl+C。你会发现，当它重新启动时，它会自动从上一步继续！

所有源代码都可以在 GitHub 上，https://github.com/dbos-inc/durable-swarm/tree/main/examples/reliable_refund 获取。

让我们开始使用OpenAI的轻量级多代理调度框架 Swarm 来创建一个AI驱动的退款代理。这个代理包含两个功能：process_refund 帮助用户退货，以及 apply_discount 为用户的未来购买应用折扣。这些功能可以根据OpenAI语言模型的输出被调用。此外，process_refund 功能会执行一些子步骤（refund_step），以此来模拟一个复杂的退款工作流程。

    从 swarm 导入 Agent
    从 dbos 导入 DBOS

    def process_refund(context_variables, item_id, reason="NOT SPECIFIED"):
        """为商品退款。请确保您有形式为 item_... 的 item_id。在处理退款前，请确认用户同意。"""
        user_name = context_variables.get("user_name", "user")
        print(f"[模拟] 正在为 {user_name} 退款，商品 {item_id}，原因：{reason}...")
        for i in range(1, 6):
            refund_step(i)
            DBOS.sleep(1)
        print("[模拟] 退款成功处理!")
        return "Success!"

    @DBOS.step()
    def refund_step(step_id):
        print(f"[模拟] 正在处理退款步骤 {step_id}... 若要退出，请按 Control + C")

    @DBOS.step()
    def apply_discount():
        """为用户的购物车应用折扣。"""
        print("[模拟] 正在应用折扣...")
        return "已应用 11% 折扣"

    refunds_agent = Agent(
        name="Refunds Agent",
        instructions="帮助用户进行退款。如果原因是价格过高，提供用户退款码。如果用户坚持，处理退款。",
        functions=[process_refund, apply_discount],
    )

进入全屏 退出全屏

我们将 refund_step 和 apply_discount 作为 DBOS 步骤集成到代理的工作流中。这样，如果代理的工作流在处理退款时中断，当它重新启动时，将从最后一个完成的步骤开始。DBOS 保证一旦代理的工作流开始运行，您一定能得到退款，绝不会重复退款或享受折扣两次！

接下来，我们来通过添加几行DBOS代码来增强Swarm的功能，使其更加健壮，能够应对任何故障。

这段代码将Swarm的主循环（run）定义为一个持久性DBOS工作流，并将每个聊天完成步骤定义为该工作流中的一个DBOS步骤。因此，如果该工作流被中断，它将跳过已完成的聊天步骤，并使用这些步骤的记录输出。

    从 dbos 导入 DBOS, DBOSConfiguredInstance
    从 swarm 导入 Swarm
    从 swarm.repl 导入 pretty_print_messages

    DBOS()

    @DBOS.dbos_class()
    class DurableSwarm(Swarm, DBOSConfiguredInstance):
        def __init__(self, client=None):
            # 初始化 Swarm 并设置客户端
            Swarm.__init__(self, client)
            # 初始化 DBOS 配置实例，指定为 openai_client
            DBOSConfiguredInstance.__init__(self, "openai_client")

        @DBOS.step()
        def get_chat_completion(self, *args, **kwargs):
            return super().get_chat_completion(*args, **kwargs)

        @DBOS.workflow()
        def run(self, *args, **kwargs):
            response = super().run(*args, **kwargs)
            # 打印响应消息
            pretty_print_messages(response.messages)
            return response

    DBOS.launch()

点击全屏模式，点击退出全屏

最后，我们来创建一个 DurableSwarm 实例对象，并利用退款代理处理退款！此脚本为你的代理创建了一个交互式的命令行界面（CLI）。

    从agents模块导入refunds_agent

    def main():
        client = DurableSwarm()
        print("连接到持久化的退款代理 💪🐝")

        user_name = input("\033[90m你叫什么名字\033[0m: \n")
        if user_name.strip() == "":
            return  # 如果没有提供名字就退出

        query = "我想为商品99申请退款，因为太贵了，而且颜色不喜欢！我想要退款，而且希望下次买东西能打折！"
        context_variables = {"user_name": user_name}

        client.run(
            agent=refunds_agent,
            messages=[{"role": "user", "content": query}],
            context_variables=context_variables,
        )

    if __name__ == "__main__":
        main()

全屏模式，退出全屏

要运行这个应用，你需要一个OpenAI开发者账户。在这里获取API密钥此处，在这里为你的账户设置支付选项此处。

将你的 API密钥设置为环境变量。

    export OPENAI_API密钥="API密钥值"

全屏 退出全屏

首先，克隆并进入该durable-swarm仓库：

``
git clone https://github.com/dbos-inc/durable-swarm.git
cd examples/reliable_refund

全屏 退出全屏

首先，创建一个虚拟环境，然后安装并配置[Swarm](https://github.com/openai/swarm/tree/main)和[DBOS](https://github.com/dbos-inc/dbos-transact-py)。请确保你的 Python 版本 >=3.10，因为 Swarm 需要这个版本。

python3 -m venv .venv # 使用python3创建虚拟环境
source .venv/bin/activate # 激活虚拟环境
pip install dbos git+https://github.com/openai/swarm.git # 安装依赖包

进入全屏模式；退出全屏模式

DBOS 需要一个 Postgres 数据库。如果你还没有的话，你可以用 Docker 启动一个 Postgres 数据库：

export PGPASSWORD=dbos
python3 start_postgres_docker.py

导出数据库密码并启动PostgreSQL Docker容器

全屏模式 退出全屏

最后，只需使用单个命令 `python3 main.py` 在虚拟环境中通过运行应用程序。你可以随时停止并重新启动这个程序，想停多少次都可以。例如：

python3 main.py

连接到耐久退款代理 💪🐝
请问你叫什么名字：
钱
[模拟] 正在为钱处理退款，商品 item_99，因为太贵，而且我不喜欢它的颜色...
[模拟] 正在处理退款步骤一... 按 Ctrl + C 可以退出
[模拟] 正在处理退款步骤二... 按 Ctrl + C 可以退出
[模拟] 正在处理退款步骤三... 按 Ctrl + C 可以退出
按 Ctrl + C 可以停止
^C⏎

python3 main.py

连接到耐久退款代理 💪🐝
[模拟] 正在为钱处理退款，商品 item_99，因为太贵，而且我不喜欢它的颜色...
[模拟] 正在处理退款步骤四... 按 Ctrl + C 可以退出
[模拟] 正在处理退款步骤五... 按 Ctrl + C 可以退出
[模拟] 退款处理成功！
[模拟] 正在应用折扣...
退款代理说：处理退款('item_id'='item_99', '原因'='太贵，我不喜欢它的颜色')
应用了折扣()
退款代理说：我已经为 item 99 办理了退款手续，并为下次购买应用了 11% 的折扣。如果您还需要其他帮助，请随时联系我！随时欢迎您的咨询！

进入全屏模式 退出全屏模式

## 下一步是:

来看看 DBOS 如何让您的应用程序更加扩展性和恢复力，试试看：

* 了解更多关于[Durable Swarm](https://github.com/dbos-inc/durable-swarm)吧
* 使用[durable execution](https://docs.dbos.dev/python/programming-guide)编写不会崩溃的工作流。
* 使用[队列](https://docs.dbos.dev/python/tutorials/queue-tutorial)优雅地处理API速率限制。
* 使用[计划工作流](https://docs.dbos.dev/python/tutorials/scheduled-workflows)在固定的时间间隔内运行您的函数。
* 想看看用DBOS能做什么吗？探索其他[示例应用程序](https://docs.dbos.dev/examples)。