第 5 课：多轮对话客服
query() 和 Client 的区别
到目前为止我们一直用 query()。它的特点是一进一出，像发邮件：你发一个问题，AI 回一个答案，完事。
但客服对话不是这样的：
客户: 我的订单有问题
AI: 请问您的订单号是多少？ ← 需要追问
客户: ORD-001
AI: [查订单] 您好，这个订单... ← 需要记住上文
客户: 那我要退货
AI: [查退货政策] 根据政策... ← 还是要记住之前的对话
这种场景需要 ClaudeSDKClient——它能维持一个持续的对话会话，有上下文记忆。
query():
用户 ──消息──→ AI ──回复──→ 结束
（每次都是全新的，不记得之前说了什么）
ClaudeSDKClient:
用户 ←──→ AI ←──→ 用户 ←──→ AI ...
（一个持续的会话，记得所有对话历史）
基础多轮对话
import asyncio
from claude_agent_sdk import (
ClaudeSDKClient, ClaudeAgentOptions,
AssistantMessage, ResultMessage, TextBlock, ToolUseBlock,
)
SYSTEM_PROMPT = """你是智能客服助手"小智"。规则：
1. 用中文、友好地和客户交流
2. 如果信息不够，主动追问
3. 每次回复不超过 3 句话
4. 遇到解决不了的问题说"我帮您转人工客服"
"""
async def main():
options = ClaudeAgentOptions(
system_prompt=SYSTEM_PROMPT,
)
print("🤖 客服小智上线（输入 quit 退出）\n")
async with ClaudeSDKClient(options=options) as client:
while True:
user_input = input("客户: ").strip()
if user_input.lower() in ['quit', 'exit', 'q']:
print("🤖 小智: 感谢咨询，祝您生活愉快！再见~")
break
if not user_input:
continue
await client.query(user_input)
async for msg in client.receive_response():
if isinstance(msg, AssistantMessage):
for block in msg.content:
if isinstance(block, TextBlock):
print(f"🤖 小智: {block.text}")
print()
asyncio.run(main)
试试这个对话：
客户: 我想退款
🤖 小智: 请问您的订单号是多少？
客户: ORD-001
🤖 小智: 请问您退款的原因是什么？
客户: 产品质量有问题
🤖 小智: 很抱歉给您带来不好的体验。我这边帮您...
AI 记得之前说了什么——这就是 Client 的多轮对话能力。
带工具的多轮对话
结合第 4 课的工具，做一个有真实数据的客服：
import asyncio
import json
from claude_agent_sdk import (
ClaudeSDKClient, ClaudeAgentOptions,
AssistantMessage, ResultMessage, TextBlock, ToolUseBlock,
tool, create_sdk_mcp_server,
)
# === 模拟数据 ===
ORDERS = {
"ORD-001": {"id": "ORD-001", "customer": "张三", "product": "基础版年付",
"amount": 1200, "status": "active", "date": "2025-01-15"},
"ORD-002": {"id": "ORD-002", "customer": "李四", "product": "企业版月付",
"amount": 500, "status": "active", "date": "2025-03-01"},
}
TICKETS = {} # 新创建的工单存这里
ticket_counter = 100
# === 工具 ===
@tool("get_order", "查询订单详情", {"order_id": {"type": "string", "description": "订单号"}})
async def get_order(order_id: str) -> str:
order = ORDERS.get(order_id.upper())
if not order:
return f"未找到订单 {order_id}。请核实订单号格式（如 ORD-001）。"
return json.dumps(order, ensure_ascii=False, indent=2)
@tool("create_ticket", "创建新的客服工单",
{"subject": {"type": "string", "description": "工单标题"},
"description": {"type": "string", "description": "问题描述"},
"priority": {"type": "string", "description": "优先级: high/medium/low"}})
async def create_ticket(subject: str, description: str, priority: str = "medium") -> str:
global ticket_counter
ticket_counter += 1
ticket_id = f"TK-{ticket_counter}"
TICKETS[ticket_id] = {
"id": ticket_id, "subject": subject,
"description": description, "priority": priority,
"status": "open",
}
return f"✅ 工单已创建: {ticket_id} | 标题: {subject} | 优先级: {priority}"
@tool("check_refund_eligibility", "检查订单是否符合退款条件",
{"order_id": {"type": "string", "description": "订单号"}})
async def check_refund_eligibility(order_id: str) -> str:
order = ORDERS.get(order_id.upper())
if not order:
return f"未找到订单 {order_id}"
# 简单模拟退款规则
from datetime import datetime, timedelta
order_date = datetime.strptime(order["date"], "%Y-%m-%d")
days = (datetime.now() - order_date).days
if days <= 7:
return f"✅ 订单 {order_id} 在 7 天内，可全额退款 ¥{order['amount']}"
elif days <= 30:
refund = order["amount"] * (30 - days) / 30
return f"⚠️ 订单 {order_id} 超过 7 天，可部分退款约 ¥{refund:.0f}"
else:
return f"❌ 订单 {order_id} 超过 30 天，不符合退款条件"
# === MCP 服务器 ===
support_server = create_sdk_mcp_server(
name="support",
version="1.0.0",
tools=[get_order, create_ticket, check_refund_eligibility],
)
# === 主程序 ===
SYSTEM_PROMPT = """你是客服助手"小智"。你有以下工具：
1. get_order - 查询订单信息
2. create_ticket - 为客户创建工单
3. check_refund_eligibility - 检查退款资格
工作规则：
- 先了解客户问题，信息不够就追问
- 需要查数据时主动使用工具
- 回复要友好、专业、简洁
- 退款/退货必须先检查资格
- 解决不了的转人工
- 用中文回复"""
async def main():
options = ClaudeAgentOptions(
system_prompt=SYSTEM_PROMPT,
mcp_servers={"support": support_server},
allowed_tools=[
"mcp__support__get_order",
"mcp__support__create_ticket",
"mcp__support__check_refund_eligibility",
],
)
print("=" * 40)
print("🤖 客服小智 v2.0（带工具版）")
print("=" * 40)
print("输入 quit 退出\n")
async with ClaudeSDKClient(options=options) as client:
while True:
user_input = input("客户: ").strip()
if user_input.lower() in ['quit', 'exit', 'q']:
print("🤖 小智: 感谢咨询，再见！")
break
if not user_input:
continue
await client.query(user_input)
async for msg in client.receive_response():
if isinstance(msg, AssistantMessage):
for block in msg.content:
if isinstance(block, TextBlock):
print(f"🤖 小智: {block.text}")
elif isinstance(block, ToolUseBlock):
print(f" 🔧 [正在{block.name}...]")
print()
asyncio.run(main)
试试这样的对话：
客户: 我想退款
🤖 小智: 请问您的订单号是多少？
客户: ORD-001
🔧 [正在 get_order...]
🔧 [正在 check_refund_eligibility...]
🤖 小智: 您的订单 ORD-001 是基础版年付 ¥1200。
由于已超过 7 天，可以部分退款约 ¥XXX。需要帮您提交退款申请吗？
客户: 好的，帮我提交
🔧 [正在 create_ticket...]
🤖 小智: 已为您创建退款工单 TK-101，我们会在 1-3 个工作日内处理。
AI 自动完成了：查订单 → 检查退款资格 → 创建工单，全程多轮对话。
对话管理进阶
中断处理
客户可能突然改变话题，用 interrupt() 处理：
async with ClaudeSDKClient(options=options) as client:
await client.query("帮我查一下所有订单的情况") # 可能很久
# 如果 AI 处理太久，可以中断
async def watch_response():
async for msg in client.receive_response():
if isinstance(msg, AssistantMessage):
for block in msg.content:
if isinstance(block, TextBlock):
print(f"🤖 {block.text}")
task = asyncio.create_task(watch_response())
# 超时中断
await asyncio.sleep(10)
print("\n⏱️ 处理时间过长，中断...")
await client.interrupt()
await task
# 发新指令
await client.query("算了，只帮我查 ORD-001 就行")
async for msg in client.receive_response():
# 处理新回复...
pass
动态调整模型
对话进行中可以切换模型——简单问题用便宜模型，复杂问题用强模型：
async with ClaudeSDKClient(options=options) as client:
# 先用 Haiku 处理简单寒暄
await client.set_model("claude-haiku-4-5")
await client.query("你好")
async for msg in client.receive_response():
display(msg)
# 遇到复杂问题切换到 Sonnet
await client.set_model("claude-sonnet-4-5")
await client.query("我想了解一下退款政策和升级方案的对比")
async for msg in client.receive_response():
display(msg)
动态调整权限
某些操作需要更高权限：
async with ClaudeSDKClient(options=options) as client:
# 默认模式：需要确认
await client.query("查一下 ORD-001")
async for msg in client.receive_response():
display(msg)
# 客户确认后，放开权限让 AI 直接操作
await client.set_permission_mode("acceptEdits")
await client.query("帮我提交退款申请")
async for msg in client.receive_response():
display(msg)
消息类型详解
在 receive_response() 中你会收到不同类型的消息：
async for msg in client.receive_response():
if isinstance(msg, AssistantMessage):
# AI 的回复，可能包含多种内容
for block in msg.content:
if isinstance(block, TextBlock):
# 文字回复 → 展示给客户
print(block.text)
elif isinstance(block, ToolUseBlock):
# AI 调用了工具 → 可以显示"正在处理..."
print(f"正在 {block.name}...")
elif isinstance(msg, UserMessage):
# 工具调用的结果（SDK 自动处理）
for block in msg.content:
if isinstance(block, ToolResultBlock):
# 工具返回的数据
pass
elif isinstance(msg, ResultMessage):
# 对话结束
print(f"费用: ${msg.total_cost_usd:.4f}")
print(f"耗时: {msg.duration_ms}ms")
本课小结
ClaudeSDKClient提供多轮对话能力，有上下文记忆- 结合 MCP 工具，AI 能在对话中随时查询数据和执行操作
interrupt()处理超时和话题切换set_model()动态切换模型平衡成本set_permission_mode()动态调整权限
课后练习
- 给客服加一个"满意度评价"功能：对话结束前问客户打几分
- 实现对话历史保存：每轮对话记录到文件中
- 加一个超时机制：如果 AI 处理超过 15 秒就自动中断并道歉