引言
我们最近发布了新的开源 Agents SDK，旨在帮助您使用轻量级、易于使用的包和最少的抽象来构建代理式人工智能应用程序。
本指南演示了如何将 Agents SDK 与 Stripe 的 API 结合使用来处理争议管理，这是许多企业面临的常见运营挑战。具体来说，我们关注两个实际场景：
-
公司失误： 公司明显犯了错误的情况，例如未能履行订单，此时接受争议是适当的操作。
-
客户争议（最终销售）： 客户明知故犯地对交易提出争议，尽管他们收到了正确的商品并理解购买是最终销售，需要进一步调查以收集支持证据。
为了解决这些场景，我们将引入三个不同的代理：
-
分类代理： 根据订单的履行状态确定是接受还是升级争议。
-
接受代理： 通过自动接受争议并提供简洁的理由来处理明确的案例。
-
调查代理： 通过分析通信记录和订单信息来收集关键证据，对争议进行彻底调查。
在本指南中，我们将逐步指导您，演示自定义代理工作流如何自动化争议管理并支持您的业务运营。
先决条件
在运行此指南之前，您必须设置好以下帐户并完成一些设置操作。这些先决条件对于与本项目中使用的 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 # 仅导入您需要的
import stripe
from typing_extensions import TypedDict, Any
# 从 .env 文件加载环境变量
load_dotenv()
# 配置日志记录
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# 从环境变量设置 Stripe API 密钥
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:
"""
返回给定电话号码的电话通话记录列表。
每条记录可能包括通话时间戳、时长、备注以及关联的 order_id（如果适用）。
"""
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:
"""
通过 ID 从预定义的订单列表中检索订单。
返回相应的订单对象或“未找到订单”。
"""
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:
"""
返回给定电子邮件地址的电子邮件记录列表。
"""
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:
"""
按 ID 检索 Stripe 支付意图。
成功时返回支付意图对象，失败时返回空字典。
"""
try:
return stripe.PaymentIntent.retrieve(payment_intent_id)
except stripe.error.StripeError as e:
logger.error(f"检索支付意图时发生 Stripe 错误：{e}")
return {}
@function_tool
async def close_dispute(dispute_id: str) -> dict:
"""
按 ID 关闭 Stripe 争议。
成功时返回争议对象，失败时返回空字典。
"""
try:
return stripe.Dispute.close(dispute_id)
except stripe.error.StripeError as e:
logger.error(f"关闭争议时发生 Stripe 错误：{e}")
return {}
定义代理
- 争议接收代理（investigator_agent）负责通过收集所有相关证据并提供报告来调查争议。
- 接受争议代理（accept_dispute_agent）通过自动关闭争议并提供简要解释来处理被确定为有效的争议。
- 分类代理（triage_agent）充当决策者，从支付意图的元数据中提取订单 ID，检索详细的订单信息，然后决定是升级争议给调查员还是将其传递给接受争议代理。
- 总之，这些代理构成了一个模块化的工作流，通过将特定任务委派给专门的代理来自动化和简化争议解决过程。
investigator_agent = Agent(
name="Dispute Intake Agent",
instructions=(
"作为争议调查员，请在您的最终输出中汇总以下详细信息：\n\n"
"争议详情：\n"
"- 争议 ID\n"
"- 金额\n"
"- 争议原因\n"
"- 卡品牌\n\n"
"支付和订单详情：\n"
"- 订单的履行状态\n"
"- 承运商和追踪号\n"
"- TOS 接受确认\n\n"
"电子邮件和电话记录：\n"
"- 任何相关的电子邮件线索（包括完整的正文文本）\n"
"- 任何相关的电话记录\n"
),
model="o3-mini",
tools=[get_emails, get_phone_logs]
)
accept_dispute_agent = Agent(
name="Accept Dispute Agent",
instructions=(
"您是负责接受争议的代理。请执行以下操作：\n"
"1. 使用提供的争议 ID 关闭争议。\n"
"2. 提供接受争议的原因的简短解释。\n"
"3. 引用从数据库检索到的任何相关订单详细信息（例如，未发货的订单等）。\n\n"
"然后，请按此确切格式生成您的最终输出：\n\n"
"争议详情：\n"
"- 争议 ID\n"
"- 金额\n"
"- 争议原因\n\n"
"订单详情：\n"
"- 订单的履行状态\n\n"
"关闭争议的原因\n"
),
model="gpt-4o",
tools=[close_dispute]
)
triage_agent = Agent(
name="Triage Agent",
instructions=(
"请执行以下操作：\n"
"1. 从支付意图的元数据中查找订单 ID。\n"
"2. 检索有关订单的详细信息（例如，发货状态）。\n"
"3. 如果订单已发货，请将此争议升级给调查代理。\n"
"4. 如果订单未发货，请接受争议。\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):
"""检索给定 PaymentIntent 的争议并处理数据。"""
disputes_list = stripe.Dispute.list(payment_intent=payment_intent_id)
if not disputes_list.data:
logger.warning("未找到 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)
# 将争议数据传递给分类代理
result = await Runner.run(triage_agent, input=event_str)
logger.info("工作流结果：%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 功能：
- 代理循环：管理工具调用，将结果传达给 LLM，并循环直到完成。
- 交接：实现多个专用代理之间的协调和任务委派。
- 函数工具：将 Python 函数转换为具有自动模式生成和验证的工具。
此外，该 SDK 还提供内置的跟踪功能，可通过 OpenAI 信息中心访问。跟踪功能可帮助您在开发和生产阶段可视化、调试和监控您的代理工作流。它还与 OpenAI 的评估、微调和蒸馏工具无缝集成。
虽然在本 Notebook 中没有直接介绍，但强烈建议在生产应用程序中实施保护栏，以验证输入并主动检测错误。
总而言之，本 Notebook 为进一步探索奠定了清晰的基础，强调了 OpenAI Agents SDK 如何促进直观有效的代理驱动工作流。