2025年3月17日

使用智能体SDK和Stripe API实现争议管理自动化

在Github中打开以Markdown格式查看

我们最近发布了全新的开源智能体SDK，旨在帮助您使用轻量级、易用且抽象程度最低的软件包来构建智能体AI应用。

本手册展示了如何结合使用智能体SDK和Stripe API来处理争议管理这一许多企业面临的常见运营挑战。具体而言，我们重点介绍两个实际应用场景：

  1. 公司失误:
    当公司明显犯错的情况，例如未能履行订单，此时接受争议是恰当的处理方式。

  2. 客户争议（最终销售）：
    一种场景，客户在收到正确商品且明知是最终销售的情况下仍故意对交易提出争议，需要进一步调查以收集支持证据。

为了解决这些场景，我们将介绍三种不同的智能体：

  • 分类智能体：
    根据订单的履行状态决定是接受还是升级争议。

  • 受理智能体:
    通过自动接受争议来处理明确案例，提供简洁的理由说明。

  • 调查员智能体:
    通过分析通信记录和订单信息进行彻底调查，收集关键证据来解决争议。

在本指南中，我们将逐步引导您，展示定制化智能体工作流如何自动化纠纷管理并支持您的业务运营。

先决条件

在运行本手册之前，您需要设置以下账户并完成一些准备工作。这些前提条件对于与本项目中使用的API进行交互至关重要。

1. OpenAI 账户

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

  • 操作步骤:
    注册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的评估、微调和蒸馏工具无缝集成。

虽然我们未在本笔记本中直接涉及，但强烈建议在生产应用中实施防护栏机制，以验证输入并主动检测错误。

总的来说，这个笔记本为进一步探索奠定了清晰的基础，重点展示了OpenAI Agents SDK如何促进直观高效的智能体驱动工作流程。