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

本教程面向有编程基础的开发者，深入讲解如何利用OpenAI Agents SDK与Stripe API构建自动化争议管理系统。通过三个智能体（分流、接受、调查）协同工作，处理公司失误和客户争议两类场景。教程涵盖环境搭建、工具函数定义、智能体配置、工作流触发及生产优化，强调模块化设计、函数工具和交接机制，帮助开发者掌握AI工程实践。

技术背景

争议管理是电商和SaaS企业常见的运营痛点，人工处理耗时且易出错。传统自动化依赖硬编码规则，难以应对复杂多变的争议场景。OpenAI Agents SDK提供轻量级智能体框架，允许开发者以Python优先的方式构建可扩展的AI工作流。结合Stripe API，可实现争议的自动分流、接受或调查，提升效率并降低运营成本。本教程适用于需要处理大量支付争议的团队，但需注意智能体决策依赖模型能力，且Stripe测试模式与生产环境存在差异。

核心原理

核心原理是利用Agents SDK的智能体循环（Agent Loop）和函数工具（Function Tools）机制。智能体接收输入（如争议数据），通过LLM推理决定调用哪些工具（如检索订单、关闭争议），工具返回结果后再次交给LLM，循环直至生成最终输出。交接（Handoffs）允许智能体将任务委派给更专业的智能体，实现模块化分工。数据流为：Stripe争议触发→分流智能体提取订单ID→根据履行状态决定接受或调查→接受智能体调用close_dispute工具关闭争议，调查智能体收集证据生成报告。关键设计决策包括：使用metadata传递订单ID、工具函数模拟外部数据、模型选择（o3-mini用于调查，gpt-4o用于决策）。

架构分析

系统采用三层架构：数据层（Stripe API、模拟订单/通信数据）、工具层（函数工具封装外部调用）、智能体层（三个智能体协作）。分流智能体作为入口，通过retrieve_payment_intent和get_order工具获取订单信息，根据履行状态决定交接给接受或调查智能体。接受智能体调用close_dispute工具自动关闭争议，调查智能体调用get_emails和get_phone_logs收集证据。设计思想强调单一职责：每个智能体专注特定任务，通过交接解耦，便于扩展和维护。可修改点包括：替换模拟数据为真实数据库、增加更多智能体（如申诉智能体）、调整模型参数。

完整实战教程

简介

我们最近发布了新的开源Agents SDK，旨在通过轻量级、易于使用的包和极简抽象，帮助您构建智能体AI应用。

本食谱演示了如何将Agents SDK与Stripe的API结合使用，以处理争议管理——这是许多企业面临的常见运营挑战。具体来说，我们关注两个现实场景：

  1. 公司失误： 公司明显出错的情况，例如未能履行订单，此时接受争议是合适的做法。
  1. 客户争议（最终销售）： 客户在收到正确商品并理解购买为最终销售的情况下，仍故意对交易提出争议，需要进一步调查以收集支持证据。

为了应对这些场景，我们将介绍三个不同的智能体：

在整个食谱中，我们将逐步指导您，展示自定义智能体工作流如何自动化争议管理并支持您的业务运营。

先决条件

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

1. OpenAI 账户

2. Stripe 账户

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")

定义函数工具

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

<br>

@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 = 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

场景1：公司失误（未收到产品）

此场景代表公司明显出错的情况——例如，未能履行或发货订单。在这种情况下，接受争议而非提出异议可能是合适的。

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 特性：

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

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

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

常见问题

常见错误：1) API密钥未正确加载，导致Stripe调用失败——检查.env文件及load_dotenv()调用。2) 支付意图未关联争议，process_dispute返回None——确保使用正确的测试卡号（如pm_card_createDispute）。3) 智能体工具调用失败，日志显示Stripe错误——检查网络和密钥权限。4) 模型输出格式不符合预期——在instructions中明确输出格式。排查方法：启用日志（logging.basicConfig(level=logging.INFO)），查看OpenAI仪表板追踪，逐步测试工具函数。生产优化：使用异步并发处理多个争议，增加重试机制，实施护栏（Guardrails）验证输入，监控智能体性能并定期评估。

企业应用扩展

可扩展方向：1) 集成真实数据库（如PostgreSQL）替换模拟数据。2) 增加更多争议处理场景（如欺诈、重复收费）。3) 使用Stripe Webhook实时触发工作流。4) 结合OpenAI评估工具优化智能体指令。5) 部署为API服务，使用FastAPI封装。6) 添加人工审核环节，当智能体置信度低时转人工。7) 利用SDK的追踪功能进行性能监控和调试。

总结

本教程面向有编程基础的开发者，深入讲解如何利用OpenAI Agents SDK与Stripe API构建自动化争议管理系统。通过三个智能体（分流、接受、调查）协同工作，处理公司失误和客户争议两类场景。教程涵盖环境搭建、工具函数定义、智能体配置、工作流触发及生产优化，强调模块化设计、函数工具和交接机制，帮助开发者掌握AI工程实践。

内容来源与验证

继续系统学习

站内还有更多完整实战教程

加入会员，解锁全部教程、完整代码与持续更新内容。

加入会员
查看全部AI 智能体教程