基于 MongoDB 的多智能体订单管理系统

本笔记本实现了一个多智能体系统，用于管理产品订单、库存和交付，使用

smolagents 用于智能体管理
MongoDB 用于数据持久化
DeepSeek Chat 作为 LLM 模型

设置 MongoDB Atlas

在 https://mongodb.ac.cn/cloud/atlas/register 创建一个免费的 MongoDB Atlas 账户
创建一个新集群 (免费层级已足够)
通过添加您的 IP 地址配置网络访问
创建一个具有读/写权限的数据库用户
从 Atlas UI 获取您的连接字符串 (点击 “Connect” > “Connect your application”)
将连接字符串中的 <password> 替换为您的数据库用户密码
在网络访问设置中启用来自您的 IP 地址的网络访问

安全注意事项

当使用 MongoDB Atlas 时

永远不要将包含凭据的连接字符串提交到版本控制
使用环境变量或安全密钥管理
将数据库用户权限限制为仅限于需要的权限
在 Atlas 网络访问设置中启用 IP 允许列表

设置

首先，让我们安装所需的依赖项

!pip install smolagents pymongo litellm

导入依赖项

在您的密钥中设置来自 https://www.deepseek.com/ 的 MONGODB_URI 和 DEEPSEEK_API_KEY (或任何其他 LLM 提供商)

导入所有必需的库并设置 LLM 模型

from smolagents.agents import ToolCallingAgent
from smolagents import tool, LiteLLMModel, ManagedAgent, CodeAgent
from pymongo import MongoClient
from datetime import datetime
from google.colab import userdata
from typing import List, Dict, Optional

# Initialize LLM model
MODEL_ID = "deepseek/deepseek-chat"
MONGODB_URI = userdata.get("MONGO_URI")
DEEPSEEK_API_KEY = userdata.get("DEEPSEEK_API_KEY")

数据库连接类

创建一个 MongoDB 连接管理器

mongoclient = MongoClient(MONGODB_URI, appname="devrel.showcase.multi-smolagents")
db = mongoclient.warehouse

智能体工具定义

我们的系统实现了三个用于仓库管理的核心工具

工作流程

Inventory Management Tools:
+-------------------+-------------------+
| Tool              | Description       |
+-------------------+-------------------+
| check_stock       | Queries stock     |
|                   | levels            |
+-------------------+-------------------+
| update_stock      | Adjusts inventory |
|                   | quantities        |
+-------------------+-------------------+

Order Management Tools:
+-------------------+-------------------+
| Tool              | Description       |
+-------------------+-------------------+
| create_order      | Creates new order |
|                   | document          |
+-------------------+-------------------+

Delivery Management Tools:
+-------------------+-------------------+
| Tool              | Description       |
+-------------------+-------------------+
| update_delivery   | Updates delivery  |
| _status           | status            |
+-------------------+-------------------+

Decision Flow:
+-------------------+-------------------+
| Step              | Action            |
+-------------------+-------------------+
| 1. Create Order   | Uses `create_order`|
|                   | tool to create    |
|                   | order document    |
+-------------------+-------------------+
| 2. Update Stock   | Uses `update_stock`|
|                   | tool to adjust    |
|                   | inventory         |
+-------------------+-------------------+
| 3. Update Delivery| Uses `update_delivery`|
| Status            | _status tool to   |
|                   | set delivery      |
|                   | status to         |
|                   | `in_transit`      |
+-------------------+-------------------+

为每种智能体类型定义工具

@tool
def check_stock(product_id: str) -> Dict:
    """Query product stock level.

    Args:
        product_id: Product identifier

    Returns:
        Dict containing product details and quantity
    """
    return db.products.find_one({"_id": product_id})


@tool
def update_stock(product_id: str, quantity: int) -> bool:
    """Update product stock quantity.

    Args:
        product_id: Product identifier
        quantity: Amount to decrease from stock

    Returns:
        bool: Success status
    """
    result = db.products.update_one({"_id": product_id}, {"$inc": {"quantity": -quantity}})
    return result.modified_count > 0

@tool
def create_order(products: any, address: str) -> str:
    """Create new order for all provided products.

    Args:
        products: List of products with quantities
        address: Delivery address

    Returns:
        str: Order ID message
    """
    order = {"products": products, "status": "pending", "delivery_address": address, "created_at": datetime.now()}
    result = db.orders.insert_one(order)
    return f"Successfully ordered : {str(result.inserted_id)}"

from bson.objectid import ObjectId


@tool
def update_delivery_status(order_id: str, status: str) -> bool:
    """Update order delivery status to in_transit once a pending order is provided

    Args:
        order_id: Order identifier
        status: New delivery status is being set to in_transit or delivered

    Returns:
        bool: Success status
    """
    if status not in ["pending", "in_transit", "delivered", "cancelled"]:
        raise ValueError("Invalid delivery status")

    result = db.orders.update_one({"_id": ObjectId(order_id), "status": "pending"}, {"$set": {"status": status}})
    return result.modified_count > 0

主订单管理系统

此类实现了一个用于订单处理的多智能体架构，包含以下组件

库存智能体: 处理库存检查和更新
订单智能体: 管理订单创建和文档记录
交付智能体: 控制订单交付状态更改
管理器智能体: 协调其他智能体之间工作流程

系统遵循以下流程

为客户请求创建订单文档
验证和更新产品库存水平
初始化交付跟踪状态
通过管理器协调智能体交互

主要特点

异步多智能体协调
自动化库存管理
订单状态跟踪
交付管道集成

定义协调所有智能体的主系统类

class OrderManagementSystem:
    """Multi-agent order management system"""

    def __init__(self, model_id: str = MODEL_ID):
        self.model = LiteLLMModel(model_id=model_id, api_key=DEEPSEEK_API_KEY)

        # Create agents
        self.inventory_agent = ToolCallingAgent(tools=[check_stock, update_stock], model=self.model, max_iterations=10)

        self.order_agent = ToolCallingAgent(tools=[create_order], model=self.model, max_iterations=10)

        self.delivery_agent = ToolCallingAgent(tools=[update_delivery_status], model=self.model, max_iterations=10)

        # Create managed agents
        self.managed_agents = [
            ManagedAgent(self.inventory_agent, "inventory", "Manages product inventory"),
            ManagedAgent(self.order_agent, "orders", "Handles order creation"),
            ManagedAgent(self.delivery_agent, "delivery", "Manages delivery status"),
        ]

        # Create manager agent
        self.manager = CodeAgent(
            tools=[],
            system_prompt="""For each order:
            1. Create the order document
            2. Update the inventory
            3. Set deliviery status to in_transit

            Use relevant agents:  {{managed_agents_descriptions}}  and you can use {{authorized_imports}}
            """,
            model=self.model,
            managed_agents=self.managed_agents,
            additional_authorized_imports=["time", "json"],
        )

    def process_order(self, orders: List[Dict]) -> str:
        """Process a set of orders.

        Args:
            orders: List of orders each has address and products

        Returns:
            str: Processing result
        """
        return self.manager.run(
            f"Process the following  {orders} as well as substract the ordered items from inventory."
            f"to be delivered to relevant addresses"
        )

添加示例数据

为了测试我们的订单管理系统，我们需要使用示例产品数据填充 MongoDB 数据库。以下部分展示了如何添加测试产品及其价格和数量。您可以修改产品详细信息或按照相同的结构添加更多商品。每个产品都有一个唯一的 ID、名称、价格和初始库存数量。

示例数据提供了一系列具有不同价格点和库存水平的电子产品，以演示库存跟踪。

为了测试系统，您可能需要向 MongoDB 添加一些示例产品

>>> def add_sample_products():
...     db.products.delete_many({})
...     sample_products = [
...         {"_id": "prod1", "name": "Laptop", "price": 999.99, "quantity": 10},
...         {"_id": "prod2", "name": "Smartphone", "price": 599.99, "quantity": 15},
...         {"_id": "prod3", "name": "Headphones", "price": 99.99, "quantity": 30},
...     ]

...     db.products.insert_many(sample_products)
...     print("Sample products added successfully!")


>>> # Uncomment to add sample products
>>> add_sample_products()

Sample products added successfully!

测试系统

这是一个测试数据方法的 Markdown 描述

测试策略概述

我们使用两种不同的订单场景进行测试
- 多产品订单 (笔记本电脑 + 智能手机)
- 单产品订单 (耳机)

测试数据设计

产品代表不同价位的常见电子产品
订单数量有意设置较小，以避免耗尽库存
多个送货地址以模拟真实世界场景

替代测试示例

批量订单: 同一产品的多个单位
混合类别订单: 高/低价值商品的组合
边缘情况: 接近库存限制的订单
无效场景: 库存不足的产品订单

该测试演示了

多产品订单处理
库存水平管理
交付状态更新
不同地点的地址处理

让我们使用示例订单测试我们的系统

>>> # Initialize system
>>> system = OrderManagementSystem()

>>> # Create test orders
>>> test_orders = [
...     {
...         "products": [{"product_id": "prod1", "quantity": 2}, {"product_id": "prod2", "quantity": 1}],
...         "address": "123 Main St",
...     },
...     {"products": [{"product_id": "prod3", "quantity": 3}], "address": "456 Elm St"},
... ]

>>> # Process order
>>> result = system.process_order(orders=test_orders)

>>> print("Orders processing result:", result)

Orders processing result: Here’s the response to your request:

---

### **Processed Orders and Inventory Update**

1. **Orders Created**:
   - **Order 1**:
     - **Products**:
       - `prod1`: 2 units
       - `prod2`: 1 unit
     - **Delivery Address**: `123 Main St`
     - **Order ID**: `677b8a9ff033af3a53c9a75a`
   - **Order 2**:
     - **Products**:
       - `prod3`: 3 units
     - **Delivery Address**: `456 Elm St`
     - **Order ID**: `677b8aa3f033af3a53c9a75c`

2. **Inventory Updated**:
   - **`prod1` (Laptop)**:
     - Initial stock: 6 units
     - Subtracted: 2 units
     - New stock: 4 units
   - **`prod2` (Smartphone)**:
     - Initial stock: 13 units
     - Subtracted: 1 unit
     - New stock: 12 units
   - **`prod3` (Headphones)**:
     - Initial stock: 24 units
     - Subtracted: 3 units
     - New stock: 21 units

3. **Delivery Status**:
   - Both orders have been marked as **"in_transit"** and are ready for delivery.

---

### **Summary**:
- The orders have been successfully processed.
- The inventory has been updated to reflect the subtracted quantities.
- The delivery status for both orders is now **"in_transit"**.

Let me know if you need further assistance! 😊

系统输出分析

系统成功完成以下关键操作

订单创建
- 并行处理多个订单
- 订单 ID 已生成并存储在 MongoDB 中
- 产品和送货地址已正确链接
库存管理
- 在订单处理前检查库存水平
- 在订单确认后减少数量
- 库存更新反映在 MongoDB 中
交付状态
- 初始状态设置为 “待处理”
- 处理后更新为 “运输中”
- 状态更改在订单文档中跟踪
数据一致性
- 所有 MongoDB 操作均以原子方式完成
- 订单详细信息准确保留
- 库存水平得到正确维护

当运行系统时，您可能会注意到智能体尝试将文本输出解释为 Python 代码。这是 CodeAgent 的预期行为，因为它试图理解和处理响应。经过几次尝试 (max_iterations=10) 后，如果仍然不成功，它将停止。

示例智能体行为

接收来自订单创建的文本输出
尝试将其解析为代码
使用不同的解释重试
最终完成工作流程

多智能体系统通过其错误处理和自我纠正机制展示了弹性的操作。虽然初始尝试可能会产生错误日志，但智能体通过迭代成功适应。最重要的是，最终状态显示了成功的订单处理和准确的库存水平更新，尽管存在任何中间错误，仍保持了数据一致性。

此行为是按设计进行的，不会影响系统的核心功能。实际的订单处理、库存更新和交付状态更改均通过 MongoDB 操作成功完成。

结论

在本笔记本中，我们使用 smolagents 和 MongoDB 成功实现了一个多智能体订单管理系统。我们定义了各种工具来管理库存、创建订单和更新交付状态。我们还创建了一个主系统类来协调这些智能体，并使用示例数据和订单测试了系统。

这种方法展示了将基于智能体的系统与强大的数据持久化解决方案 (如 MongoDB) 相结合以创建可扩展且高效的订单管理系统的强大功能。

< > 在 GitHub 上更新

开源 AI 食谱