使用 Agents SDK 和 Stripe API 自动化争议管理

2025 年 3 月 17 日
在 Github 中打开

我们最近发布了新的开源 **Agents SDK**，旨在帮助您使用轻量级、易于使用的软件包和最少的抽象构建代理 AI 应用程序。

本指南演示了如何结合使用 Agents SDK 和 Stripe 的 API 来处理争议管理，这是许多企业面临的常见运营挑战。具体而言，我们重点关注两种实际场景：

  1. 公司失误
    公司明显犯错的场景，例如未能履行订单，接受争议是适当的做法。

  2. 客户争议（最终销售）
    客户明知故犯地对交易提出争议的场景，尽管他们收到了正确的商品并理解购买是最终销售，需要进一步调查以收集支持证据。

为了解决这些场景，我们将介绍三个不同的代理：

  • 分诊代理
    根据订单的履行状态确定是接受还是升级争议。

  • 接受代理
    通过自动接受争议来处理明确的案例，并提供简明的理由。

  • 调查员代理
    通过分析沟通记录和订单信息来彻底调查争议，以收集必要的证据。

在本指南中，我们将逐步指导您，说明自定义代理工作流程如何自动化争议管理并支持您的业务运营。

先决条件

在运行本指南之前，您必须设置以下帐户并完成一些设置操作。这些先决条件对于与本项目中使用的 API 进行交互至关重要。

1. OpenAI 帐户

  • 目的
    您需要一个 OpenAI 帐户才能访问语言模型并使用本指南中介绍的 Agents SDK。

  • 操作
    如果您还没有 OpenAI 帐户，请注册一个 OpenAI 帐户。拥有帐户后，访问 OpenAI API 密钥页面 创建 API 密钥。

2. Stripe 帐户

  • 目的
    需要一个 Stripe 帐户来模拟支付处理、管理争议以及与 Stripe API 交互，作为我们的演示工作流程的一部分。

  • 操作
    访问 Stripe 注册页面 创建一个免费的 Stripe 帐户。

  • 查找您的 API 密钥
    登录您的 Stripe 仪表板，导航至**开发者 > API 密钥**。

  • 使用测试模式
    对所有开发和测试使用您的**测试密钥**。

3. 创建一个包含您的 OpenAI API 和 Stripe API 密钥的 .env 文件

OPENAI_API_KEY=
STRIPE_SECRET_KEY=

环境设置

首先，我们将安装必要的依赖项，然后导入库并编写一些实用函数，我们将在稍后使用它们。

%pip install python-dotenv --quiet
%pip install openai-agents --quiet
%pip install stripe --quiet
%pip install typing_extensions --quiet
import os
import logging
import json
from dotenv import load_dotenv
from agents import Agent, Runner, function_tool  # Only import what you need
import stripe
from typing_extensions import TypedDict, Any
# Load environment variables from .env file
load_dotenv()

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Set Stripe API key from environment variables
stripe.api_key = os.getenv("STRIPE_SECRET_KEY")

定义函数工具

本节定义了几个辅助函数工具，用于支持争议处理工作流程。

  • get_order、get_phone_logs 和 get_emails 通过返回基于提供的标识符的订单详细信息以及电子邮件/电话记录来模拟外部数据查找。
  • retrieve_payment_intent 与 Stripe API 交互以获取付款意图详细信息。
  • close_dispute 使用提供的争议 ID 自动关闭 Stripe 争议，确保争议得到妥善解决和记录。
@function_tool
def get_phone_logs(phone_number: str) -> list:
    """
    Return a list of phone call records for the given phone number.
    Each record might include call timestamps, durations, notes, 
    and an associated order_id if applicable.
    """
    phone_logs = [
        {
            "phone_number": "+15551234567",
            "timestamp": "2023-03-14 15:24:00",
            "duration_minutes": 5,
            "notes": "Asked about status of order #1121",
            "order_id": 1121
        },
        {
            "phone_number": "+15551234567",
            "timestamp": "2023-02-28 10:10:00",
            "duration_minutes": 7,
            "notes": "Requested refund for order #1121, I told him we were unable to refund the order because it was final sale",
            "order_id": 1121
        },
        {
            "phone_number": "+15559876543",
            "timestamp": "2023-01-05 09:00:00",
            "duration_minutes": 2,
            "notes": "General inquiry; no specific order mentioned",
            "order_id": None
        },
    ]
    return [
        log for log in phone_logs if log["phone_number"] == phone_number
    ]


@function_tool
def get_order(order_id: int) -> str:
    """
    Retrieve an order by ID from a predefined list of orders.
    Returns the corresponding order object or 'No order found'.
    """
    orders = [
        {
            "order_id": 1234,
            "fulfillment_details": "not_shipped"
        },
        {
            "order_id": 9101,
            "fulfillment_details": "shipped",
            "tracking_info": {
                "carrier": "FedEx",
                "tracking_number": "123456789012"
            },
            "delivery_status": "out for delivery"
        },
        {
            "order_id": 1121,
            "fulfillment_details": "delivered",
            "customer_id": "cus_PZ1234567890",
            "customer_phone": "+15551234567",
            "order_date": "2023-01-01",
            "customer_email": "customer1@example.com",
            "tracking_info": {
                "carrier": "UPS",
                "tracking_number": "1Z999AA10123456784",
                "delivery_status": "delivered"
            },
            "shipping_address": {
                "zip": "10001"
            },
            "tos_acceptance": {
                "date": "2023-01-01",
                "ip": "192.168.1.1"
            }
        }
    ]
    for order in orders:
        if order["order_id"] == order_id:
            return order
    return "No order found"


@function_tool
def get_emails(email: str) -> list:
    """
    Return a list of email records for the given email address.
    """
    emails = [
        {
            "email": "customer1@example.com",
            "subject": "Order #1121",
            "body": "Hey, I know you don't accept refunds but the sneakers don't fit and I'd like a refund"
        },
        {
            "email": "customer2@example.com",
            "subject": "Inquiry about product availability",
            "body": "Hello, I wanted to check if the new model of the smartphone is available in stock."
        },
        {
            "email": "customer3@example.com",
            "subject": "Feedback on recent purchase",
            "body": "Hi, I recently purchased a laptop from your store and I am very satisfied with the product. Keep up the good work!"
        }
    ]
    return [email_data for email_data in emails if email_data["email"] == email]


@function_tool
async def retrieve_payment_intent(payment_intent_id: str) -> dict:
    """
    Retrieve a Stripe payment intent by ID.
    Returns the payment intent object on success or an empty dictionary on failure.
    """
    try:
        return stripe.PaymentIntent.retrieve(payment_intent_id)
    except stripe.error.StripeError as e:
        logger.error(f"Stripe error occurred while retrieving payment intent: {e}")
        return {}

@function_tool
async def close_dispute(dispute_id: str) -> dict:
    """
    Close a Stripe dispute by ID. 
    Returns the dispute object on success or an empty dictionary on failure.
    """
    try:
        return stripe.Dispute.close(dispute_id)
    except stripe.error.StripeError as e:
        logger.error(f"Stripe error occurred while closing dispute: {e}")
        return {}

定义代理

  • **争议接收代理（investigator_agent）**负责通过收集所有相关证据并提供报告来调查争议。
  • **接受争议代理（accept_dispute_agent）**处理确定为有效的争议，方法是自动关闭争议并为该决定提供简要说明。
  • **分诊代理（triage_agent）**充当决策者，通过从付款意图的元数据中提取订单 ID，检索详细的订单信息，然后决定是将争议升级给调查员还是将其传递给接受争议代理。
  • 这些代理共同构成了一个模块化的工作流程，通过将特定任务委派给专业的代理来自动化和简化争议解决流程。
investigator_agent = Agent(
    name="Dispute Intake Agent",
    instructions=(
        "As a dispute investigator, please compile the following details in your final output:\n\n"
        "Dispute Details:\n"
        "- Dispute ID\n"
        "- Amount\n"
        "- Reason for Dispute\n"
        "- Card Brand\n\n"
        "Payment & Order Details:\n"
        "- Fulfillment status of the order\n"
        "- Shipping carrier and tracking number\n"
        "- Confirmation of TOS acceptance\n\n"
        "Email and Phone Records:\n"
        "- Any relevant email threads (include the full body text)\n"
        "- Any relevant phone logs\n"
    ),
    model="o3-mini",
    tools=[get_emails, get_phone_logs]
)


accept_dispute_agent = Agent(
    name="Accept Dispute Agent",
    instructions=(
        "You are an agent responsible for accepting disputes. Please do the following:\n"
        "1. Use the provided dispute ID to close the dispute.\n"
        "2. Provide a short explanation of why the dispute is being accepted.\n"
        "3. Reference any relevant order details (e.g., unfulfilled order, etc.) retrieved from the database.\n\n"
        "Then, produce your final output in this exact format:\n\n"
        "Dispute Details:\n"
        "- Dispute ID\n"
        "- Amount\n"
        "- Reason for Dispute\n\n"
        "Order Details:\n"
        "- Fulfillment status of the order\n\n"
        "Reasoning for closing the dispute\n"
    ),
    model="gpt-4o",
    tools=[close_dispute]
)

triage_agent = Agent(
    name="Triage Agent",
    instructions=(
        "Please do the following:\n"
        "1. Find the order ID from the payment intent's metadata.\n"
        "2. Retrieve detailed information about the order (e.g., shipping status).\n"
        "3. If the order has shipped, escalate this dispute to the investigator agent.\n"
        "4. If the order has not shipped, accept the dispute.\n"
    ),
    model="gpt-4o",
    tools=[retrieve_payment_intent, get_order],
    handoffs=[accept_dispute_agent, investigator_agent],
)

检索争议并启动代理工作流程

此函数使用提供的 payment_intent_id 从 Stripe 检索争议详细信息，并通过将检索到的争议信息传递给指定的 triage_agent 来启动争议处理工作流程。

async def process_dispute(payment_intent_id, triage_agent):
    """Retrieve and process dispute data for a given PaymentIntent."""
    disputes_list = stripe.Dispute.list(payment_intent=payment_intent_id)
    if not disputes_list.data:
        logger.warning("No dispute data found for PaymentIntent: %s", payment_intent_id)
        return None
    
    dispute_data = disputes_list.data[0]
    
    relevant_data = {
        "dispute_id": dispute_data.get("id"),
        "amount": dispute_data.get("amount"),
        "due_by": dispute_data.get("evidence_details", {}).get("due_by"),
        "payment_intent": dispute_data.get("payment_intent"),
        "reason": dispute_data.get("reason"),
        "status": dispute_data.get("status"),
        "card_brand": dispute_data.get("payment_method_details", {}).get("card", {}).get("brand")
    }
    
    event_str = json.dumps(relevant_data)
    # Pass the dispute data to the triage agent
    result = await Runner.run(triage_agent, input=event_str)
    logger.info("WORKFLOW RESULT: %s", result.final_output)
    
    return relevant_data, result.final_output
payment = stripe.PaymentIntent.create(
  amount=2000,
  currency="usd",
  payment_method = "pm_card_createDisputeProductNotReceived",
  confirm=True,
  metadata={"order_id": "1234"},
  off_session=True,
  automatic_payment_methods={"enabled": True},
)
relevant_data, triage_result = await process_dispute(payment.id, triage_agent)

场景 2：客户争议（最终销售）

此场景描述了客户故意对交易提出争议的情况，尽管他们收到了正确的产品并且完全清楚购买被明确标记为“最终销售”（不退款或退货）。此类争议通常需要进一步调查以收集证据，以便有效地对抗争议。

payment = stripe.PaymentIntent.create(
  amount=2000,
  currency="usd",
  payment_method = "pm_card_createDispute",
  confirm=True,
  metadata={"order_id": "1121"},
  off_session=True,
  automatic_payment_methods={"enabled": True},
)
relevant_data, triage_result = await process_dispute(payment.id, triage_agent)

结论

在本 Jupyter Notebook 中，我们探索了 **OpenAI Agents SDK** 的功能，演示了如何使用简单、Python 优先的方法高效创建基于代理的 AI 应用程序。具体而言，我们展示了以下 SDK 功能：

  • **代理循环**：管理工具调用，将结果传达给 LLM，并循环直到完成。
  • **移交**：支持多个专业代理之间的协调和委派任务。
  • **函数工具**：将 Python 函数转换为工具，具有自动模式生成和验证功能。

此外，SDK 还提供内置的**跟踪**功能，可通过 OpenAI 仪表板访问。跟踪功能可帮助您在开发和生产阶段可视化、调试和监控您的代理工作流程。它还可以与 OpenAI 的评估、微调和蒸馏工具顺利集成。

虽然我们没有在本 notebook 中直接介绍，但强烈建议在生产应用程序中实施**护栏**，以验证输入并主动检测错误。

总的来说，本 notebook 为进一步探索奠定了清晰的基础，强调了 OpenAI Agents SDK 如何促进直观且有效的代理驱动工作流程。