phone-agent¶
第9章 · 多模态与实时交互 · 配套项目
chapter9/phone-agent
项目说明¶
实验 9-2：使用 PineClaw Voice API 构建电话 Agent¶
配套《深入理解 AI Agent》第 9 章实验 9-2。
目的¶
真实世界里很多 Agent 任务离不开拨打真实电话——联系客服协商账单、预约餐厅、 确认订单。本实验演示语音 Agent 的一个重要应用方向：Agent 不仅能与用户语音对话， 还能代替用户与外部世界进行电话交互。
上层是一个标准的 ReAct Agent：接到一个自然语言任务（如"打电话给宽带客服，
查询本月账单为何多扣了 50 元并要求解释"），它自己想清楚要拨的号码、通话目标和
上下文，调用 make_phone_call 工具完成整段通话，读取返回的结构化通话记录，
必要时追问/再拨，最后向用户汇报结果。
电话语音 API 的抽象¶
生产级电话语音 API（如 PineClaw Voice API，作者团队开发） 把一整通电话封装成一次工具调用：
你只提供三样东西——号码、目标、上下文——它的语音 Agent 就会自动完成：
- 拨号：接通被叫方；
- IVR 导航：应对"查询请按 1，转人工请按 0"这类按键菜单；
- 多轮对话：接通人工后围绕目标交涉、追问、确认关键信息；
- 转录：把整段通话转成文字。
最后返回一份结构化通话记录，而不是一段裸录音。这正是它能塞进 ReAct 循环的原因：
Agent 拿到的是结构化字段（是否达成目标、抽取的关键信息、逐轮 transcript），可以直接
据此决策与汇报。本实验返回体的形状（见 pine_voice.py 的 CallRecord）：
| 字段 | 含义 |
|---|---|
call_id |
通话唯一 ID |
phone_number / goal |
本次通话的号码与目标 |
status / goal_achieved |
通话状态 / 是否达成目标 |
duration_seconds |
通话时长 |
summary |
一句话通话摘要 |
key_fields |
抽取的关键信息（扣费原因 / 金额 / 确认号 / 时间…） |
transcript |
逐轮对话 [{speaker, text}, ...] |
follow_up_needed / follow_up_reason |
是否仍需追问及原因 |
关于 mock（重要）¶
真实 PineClaw Voice API 需要 PINECLAW_API_KEY 并会拨打真实电话号码。为便于离线跑通，
本实验用一个本地模拟客户端替代真实 API（pine_voice.py）：
- 它不接触真实电话网络，也不需要 PineClaw key；
make_phone_call内部用 OpenAI 扮演被叫方——先当自动 IVR 语音菜单，被"转人工" 后再扮演人工客服——与去电的语音 Agent 进行一段多轮对话（模拟 IVR 导航 + 客服应答）， 然后把 transcript 归纳成上表的结构化记录；- 关键在于：模拟客户端与真实 API 的输入/输出契约完全一致，因此上层 ReAct Agent 的代码在切换到真实 PineClaw SDK 时几乎无需改动。
所以本实验里出现的"扣费原因""确认号"等都是模型即时编造的模拟情节，仅用于演示 数据流，不代表任何真实通话。
真实接入 PineClaw¶
把 agent.py 里对模拟 make_phone_call 的调用替换为真实 SDK 即可，其余逻辑不变：
# pip install pine-voice
from pine_voice import PineVoiceClient # 真实 SDK（示意）
client = PineVoiceClient(api_key=os.environ["PINECLAW_API_KEY"])
def make_phone_call(phone_number, goal, context=""):
call = client.calls.create(to=phone_number, goal=goal, context=context)
result = call.wait() # 阻塞直到通话结束（可能是分钟级到小时级）
return result.to_dict() # 返回同形状的结构化通话记录
真实使用请以 PineClaw 官方文档为准；建议先拨打自己的手机验证连通性。
运行¶
cd chapter9/phone-agent
pip install -r requirements.txt
cp env.example .env
# 编辑 .env，填入 OPENROUTER_API_KEY 或 OPENAI_API_KEY（至少其一）
python demo.py
python demo.py --task "帮我打电话给餐厅订今晚 7 点 4 人的位子" # 自定义电话任务
python demo.py --dry-run # 离线跑通，无需任何 API Key
python demo.py --help # 查看全部参数
命令行参数（python demo.py --help 有中文说明）：
| 参数 | 作用 |
|---|---|
--task |
自定义电话任务（自然语言）。默认用书中的宽带账单示例 |
--phone |
可选：对方电话号码。作为已知信息交给 Agent（dry-run 下直接用作被叫号码） |
--goal |
可选：明确的通话目标。作为已知信息交给 Agent（dry-run 下直接用作通话目标） |
--model |
可选：覆盖模型（默认取 OPENAI_MODEL，即 gpt-5.6-luna） |
--dry-run |
离线脚本模式：不联网、不需要任何 API Key，仅演示 ReAct 循环与数据契约的形状 |
demo.py 会真实调用 OpenAI（除非加 --dry-run），打印三段内容：
(a) ReAct Agent 的轨迹（思考 + 发起 make_phone_call）；
(b) 返回的结构化通话记录（多轮 transcript + 是否达成目标 + 关键字段）；
(c) Agent 基于通话结果向用户的最终汇报。
模型与回退：默认聊天模型
gpt-5.6-luna。解析优先级为OPENAI_API_KEY（可选OPENAI_BASE_URL指向兼容网关，如 Moonshotkimi-k3/ 火山方舟） >OPENROUTER_API_KEY（自动把模型映射为openai/gpt-5.6-luna等provider/model形式）。 由于gpt-5.6*直连 OpenAI 需组织实名认证，推荐用 OpenRouter；只需在.env里填OPENROUTER_API_KEY即可（此时不要设OPENAI_API_KEY，否则会优先直连）。
两级「模拟」的区别¶
本实验里有两层各自独立的模拟，别混淆：
- 默认（mock）：
pine_voice.py替换掉真实电话网络，但 ReAct Agent 与被叫方对话 仍由 OpenAI 实时生成——所以需要OPENAI_API_KEY，每次对话/字段都不同。 --dry-run（离线脚本）：连 LLM 也不调用，make_phone_call直接返回一份固定脚本的 结构化通话记录。用于在没有任何 API Key、完全离线时也能把整条 ReAct 循环 （思考 → 调用工具 → 读结构化记录 → 汇报）跑通、看清其形状。脚本里的确认号由目标哈希派生， 可复现，不代表任何真实通话。
预期输出示例（真实节选）¶
[Agent 调用工具 make_phone_call] 入参:
phone_number = 10010
goal = 查询本月宽带账单多扣的50元原因，并要求处理误扣。
context = 宽带账号 hz-88231
[PineClaw 返回结构化通话记录]
状态 : completed | 是否达成目标: True
摘要 : 用户成功查询到宽带账单多扣50元的原因并申请了退款。
关键字段(key_fields):
- 扣费原因: 系统自动调整套餐费用
- 涉及金额: 50元
- 确认号: RW20231015
- 处理结果: 退款申请已成功提交
通话转录(transcript):
<< [被叫方] 欢迎致电客服热线！账单查询请按 1，业务办理请按 2，人工服务请按 0。
>> [语音Agent] 我按 0 转人工。
<< [被叫方] 您好，我是客服代表，工号12345，请问有什么可以帮助您的？
>> [语音Agent] 你好，我发现宽带账单多扣了50元，能帮我查一下原因吗？账号是 hz-88231。
...（多轮交涉、核对、给出确认号）...
Agent 向用户的最终汇报：
已成功拨打客服热线并查询原因：扣费原因=系统自动调整套餐费用；已提交退款，确认号 RW20231015。
注意：IVR 菜单、工号、确认号等都是模型即时编造的模拟情节（见「关于 mock」）， 仅演示数据流；每次运行的对话与字段会不同，但「IVR 导航 → 转人工多轮交涉 → 结构化 记录 → Agent 汇报」的形状稳定复现。
文件说明¶
| 文件 | 作用 |
|---|---|
pine_voice.py |
PineClaw Voice API 的本地模拟客户端，提供 make_phone_call 工具 |
agent.py |
把 make_phone_call 当工具的 ReAct Agent（OpenAI function calling） |
demo.py |
端到端演示：一个电话任务从下达到汇报 |
requirements.txt / env.example |
依赖与环境变量模板 |
源代码¶
agent.py¶
"""
电话 Agent —— 一个把 PineClaw Voice（make_phone_call）当作工具的 ReAct Agent。
上层是标准的 ReAct 循环（OpenAI function calling 实现）：
用户给一个任务（如"打电话给宽带客服，问清本月为何多扣 50 元并要求解释"）
-> Agent 思考需要哪些参数（号码 / 目标 / 上下文）
-> 调用 make_phone_call 工具完成整段通话
-> 读取返回的【结构化通话记录】
-> 若信息不足则追问/再拨，否则向用户给出最终汇报
工具 make_phone_call 由 pine_voice 模块提供（对真实 PineClaw Voice API 的本地模拟）。
"""
from __future__ import annotations
import json
import re
from typing import Any, Callable
# 复用 pine_voice 里的统一 client/模型解析（OPENAI_API_KEY 直连，缺失则回退 OpenRouter）。
from pine_voice import make_phone_call, _get_client, default_model
# 最多允许 Agent 发起几次工具调用（含追问/再拨），防止死循环。
_MAX_STEPS = 6
_SYSTEM_PROMPT = (
"你是一名『电话助理』Agent。用户会交给你一个需要打电话才能完成的任务，"
"你要代表用户把电话打好并汇报结果。\n\n"
"你有一个工具 make_phone_call，它会把整通电话（拨号、IVR 菜单导航、与客服多轮对话、"
"转录）全部交给 PineClaw 语音 Agent 完成，并返回结构化通话记录。\n\n"
"工作方式（ReAct）：\n"
"1. 先想清楚要拨打的号码、通话目标、需要带上的上下文（账号/姓名/已知信息）。\n"
"2. 调用 make_phone_call 完成通话。\n"
"3. 读取返回的结构化记录（goal_achieved / key_fields / summary / transcript）。\n"
"4. 若目标未达成或信息不足（follow_up_needed 为真），可以带着更明确的目标再拨一次"
"（但总次数有限，不要无谓重拨）。\n"
"5. 目标达成后，用简洁中文向用户汇报：结论 + 关键信息（金额/原因/确认号/时间等）"
"+ 后续建议（如有）。\n\n"
"注意：如果任务里连电话号码都没有，就用一个合理的占位号码并在汇报中说明。"
"始终基于工具真实返回的通话记录来汇报，不要编造通话没有提到的信息。"
)
_TOOLS = [
{
"type": "function",
"function": {
"name": "make_phone_call",
"description": (
"代表用户拨打一通真实电话。PineClaw 语音 Agent 会完成拨号、IVR 菜单导航、"
"与对方多轮对话，并返回结构化通话记录（含 transcript、是否达成目标、"
"抽取的关键字段）。"
),
"parameters": {
"type": "object",
"properties": {
"phone_number": {
"type": "string",
"description": "被叫电话号码，如 10000 或 400-810-xxxx。",
},
"goal": {
"type": "string",
"description": "本次通话要达成的明确目标（一句话）。",
},
"context": {
"type": "string",
"description": "辅助上下文：账号、姓名、已知金额、时间等，可为空。",
},
},
"required": ["phone_number", "goal"],
},
},
}
]
def run_agent(
task: str,
on_event: Callable[[str, Any], None] | None = None,
*,
model: str | None = None,
phone_hint: str | None = None,
goal_hint: str | None = None,
dry_run: bool = False,
) -> str:
"""
运行 ReAct 电话 Agent。
参数：
task: 用户的自然语言任务。
on_event: 可选回调，用于外部观察 Agent 轨迹。
事件类型：'think'(Agent 思考文本) / 'call'(工具入参) /
'record'(结构化通话记录) / 'final'(最终汇报)。
model: 覆盖使用的模型（默认取环境变量 OPENAI_MODEL）。
phone_hint: 可选的电话号码；给定时作为已知信息交给 Agent（dry-run 下直接用作被叫号码）。
goal_hint: 可选的通话目标；给定时作为已知信息交给 Agent（dry-run 下直接用作通话目标）。
dry_run: 若为真，走完全离线的脚本化 ReAct 轨迹（不联网、不需要任何 API Key）。
返回：
Agent 面向用户的最终汇报文本。
"""
def emit(kind: str, payload: Any) -> None:
if on_event:
on_event(kind, payload)
if dry_run:
return _run_agent_dryrun(task, emit, phone_hint, goal_hint)
model = model or default_model()
# 若显式给了号码/目标，就作为「已知信息」拼进用户消息，Agent 仍自行决定如何使用。
hints = []
if phone_hint:
hints.append(f"已知对方电话号码：{phone_hint}")
if goal_hint:
hints.append(f"用户明确的通话目标：{goal_hint}")
user_content = task if not hints else task + "\n\n（" + "；".join(hints) + "）"
messages: list[dict[str, Any]] = [
{"role": "system", "content": _SYSTEM_PROMPT},
{"role": "user", "content": user_content},
]
for _ in range(_MAX_STEPS):
resp = _get_client().chat.completions.create(
model=model,
messages=messages,
tools=_TOOLS,
tool_choice="auto",
temperature=0.3,
)
msg = resp.choices[0].message
# 附上模型这一步返回的 assistant 消息（可能同时含思考文本与工具调用）。
messages.append(msg.model_dump(exclude_none=True))
if msg.content:
emit("think", msg.content)
if not msg.tool_calls:
# 没有工具调用 = Agent 给出了最终答复。
final = msg.content or ""
emit("final", final)
return final
# 逐个执行工具调用（本例只有 make_phone_call）。
for tc in msg.tool_calls:
if tc.function.name != "make_phone_call":
result = {"error": f"未知工具 {tc.function.name}"}
else:
args = json.loads(tc.function.arguments or "{}")
emit("call", args)
result = make_phone_call(
phone_number=args.get("phone_number", "10000"),
goal=args.get("goal", task),
context=args.get("context", ""),
model=model,
)
emit("record", result)
messages.append(
{
"role": "tool",
"tool_call_id": tc.id,
"name": "make_phone_call",
"content": json.dumps(result, ensure_ascii=False),
}
)
# 兜底：步数用尽仍未收敛，逼模型直接总结一次。
messages.append(
{
"role": "user",
"content": "请根据以上通话记录，立即给用户一份最终汇报，不要再打电话了。",
}
)
resp = _get_client().chat.completions.create(
model=model, messages=messages, temperature=0.3
)
final = resp.choices[0].message.content or ""
emit("final", final)
return final
def _guess_phone(task: str) -> str | None:
"""从任务文本里粗略抽取一个像电话号码的数字串（>=4 位，避开『50 元』这类金额）。"""
for m in re.finditer(r"\d{4,}", task):
return m.group(0)
return None
def _guess_context(task: str) -> str:
"""从任务文本里粗略抽取账号等上下文（仅用于 dry-run 的启发式，不求完备）。"""
m = re.search(r"账[号户][^A-Za-z0-9]{0,3}([A-Za-z0-9][A-Za-z0-9\-]{2,})", task)
if m:
return f"账号 {m.group(1)}"
return ""
def _run_agent_dryrun(
task: str,
emit: Callable[[str, Any], None],
phone_hint: str | None,
goal_hint: str | None,
) -> str:
"""
完全离线的脚本化 ReAct 轨迹：不调用任何 LLM/电话 API，仅演示循环的形状——
思考(推断参数) → 行动(make_phone_call, dry_run) → 观察(结构化记录) → 汇报。
"""
phone = phone_hint or _guess_phone(task) or "10010"
goal = goal_hint or task
context = _guess_context(task)
emit(
"think",
"这是一个需要打电话才能完成的任务。我先确定通话三要素——"
f"号码={phone}；目标={goal}；上下文={context or '（无）'}。参数已足够，现在发起通话。",
)
call_args = {"phone_number": phone, "goal": goal, "context": context}
emit("call", call_args)
record = make_phone_call(**call_args, dry_run=True)
emit("record", record)
kf = record.get("key_fields", {}) or {}
kf_text = "；".join(f"{k}={v}" for k, v in kf.items()) or "（无）"
final = (
f"已（dry-run 脚本模拟）拨打 {phone} 并完成通话。\n"
f"结论：{record.get('summary', '')}\n"
f"关键信息：{kf_text}。"
)
emit("final", final)
return final
demo.py¶
"""
实验 9-2 演示：用 ReAct Agent + PineClaw Voice（模拟）完成一个电话任务。
运行：
python demo.py
它会真实调用 OpenAI：一边驱动上层 ReAct Agent 决策，一边在 make_phone_call 内部
用 OpenAI 扮演被叫方（IVR + 客服）完成一整段多轮通话，最后打印：
(a) Agent 的 ReAct 轨迹（思考 + 发起 make_phone_call）
(b) 返回的结构化通话记录（多轮 transcript + 是否达成目标 + 关键字段）
(c) Agent 基于通话结果向用户的最终汇报
"""
from __future__ import annotations
import argparse
import os
import sys
try:
from dotenv import load_dotenv
load_dotenv()
except ImportError:
pass
from agent import run_agent
def _hr(title: str = "") -> None:
line = "─" * 72
if title:
print(f"\n{line}\n{title}\n{line}")
else:
print(line)
def _print_record(rec: dict) -> None:
print(f" call_id : {rec['call_id']}")
print(f" 被叫号码 : {rec['phone_number']}")
print(f" 状态 : {rec['status']} | 是否达成目标: {rec['goal_achieved']}")
print(f" 通话时长(模拟) : {rec['duration_seconds']} 秒")
print(f" 摘要 : {rec['summary']}")
print(" 关键字段(key_fields):")
if rec["key_fields"]:
for k, v in rec["key_fields"].items():
print(f" - {k}: {v}")
else:
print(" （无）")
print(f" 需要追问 : {rec['follow_up_needed']} {rec.get('follow_up_reason', '')}")
print(" 通话转录(transcript):")
for turn in rec["transcript"]:
speaker = turn["speaker"]
# 简单对齐：语音Agent 用 >>，被叫方用 <<
arrow = ">>" if speaker == "语音Agent" else "<<"
print(f" {arrow} [{speaker}] {turn['text']}")
_DEFAULT_TASK = (
"帮我打电话给宽带客服（客服热线 10010），查询本月账单为什么多扣了 50 元，"
"要求对方解释清楚原因，如果是误扣就请他们处理。我的宽带账号是 hz-88231。"
)
def parse_args() -> argparse.Namespace:
p = argparse.ArgumentParser(
description="实验 9-2：把 PineClaw Voice（make_phone_call）当工具的 ReAct 电话 Agent。"
"给一个自然语言电话任务，Agent 自行决定号码/目标/上下文并（模拟）拨打，"
"读取结构化通话记录后向用户汇报。",
epilog="示例：\n"
" python demo.py # 书中默认的宽带账单任务（需 OPENAI_API_KEY）\n"
" python demo.py --dry-run # 完全离线：脚本化 ReAct 轨迹，无需任何 API Key\n"
" python demo.py --task \"帮我打电话给餐厅订今晚 7 点 4 人的位子\" --phone 021-8888\n"
" python demo.py --model gpt-5.6-luna # 覆盖模型",
formatter_class=argparse.RawDescriptionHelpFormatter,
)
p.add_argument("--task", default=_DEFAULT_TASK,
help="自定义电话任务（自然语言）。默认用书中的宽带账单示例。")
p.add_argument("--phone", default=None, metavar="号码",
help="可选：对方电话号码。给定时作为已知信息交给 Agent（dry-run 下直接用作被叫号码）。")
p.add_argument("--goal", default=None, metavar="目标",
help="可选：明确的通话目标。给定时作为已知信息交给 Agent（dry-run 下直接用作通话目标）。")
p.add_argument("--model", default=None, metavar="模型",
help="可选：覆盖使用的模型（默认取环境变量 OPENAI_MODEL，即 gpt-5.6-luna）。")
p.add_argument("--dry-run", action="store_true",
help="离线脚本模式：不联网、不需要任何 API Key，仅演示 ReAct 循环与数据契约的形状。")
return p.parse_args()
def main() -> None:
args = parse_args()
if not args.dry_run and not (os.getenv("OPENAI_API_KEY") or os.getenv("OPENROUTER_API_KEY")):
print("错误：未检测到 OPENAI_API_KEY 或 OPENROUTER_API_KEY。请复制 env.example 为 .env "
"并至少填入其一，或改用 python demo.py --dry-run 走完全离线的脚本演示。")
sys.exit(1)
# 书中示例任务：注意这里只给了自然语言任务，Agent 需自行决定通话参数。
task = args.task
_hr("用户任务")
print(task)
if args.dry_run:
print("\n[模式] dry-run 离线脚本模拟：以下轨迹与通话记录均为固定脚本，"
"不调用任何 LLM/电话 API，仅用于演示 ReAct 循环的形状。")
_hr("ReAct Agent 轨迹")
def on_event(kind: str, payload) -> None:
if kind == "think":
print(f"\n[Agent 思考] {payload}")
elif kind == "call":
print("\n[Agent 调用工具 make_phone_call] 入参:")
print(f" phone_number = {payload.get('phone_number')}")
print(f" goal = {payload.get('goal')}")
print(f" context = {payload.get('context', '')}")
elif kind == "record":
print("\n[PineClaw 返回结构化通话记录]")
_print_record(payload)
elif kind == "final":
pass # 最终汇报单独打印
final = run_agent(
task,
on_event=on_event,
model=args.model,
phone_hint=args.phone,
goal_hint=args.goal,
dry_run=args.dry_run,
)
_hr("Agent 向用户的最终汇报")
print(final)
print()
if __name__ == "__main__":
main()
pine_voice.py¶
"""
pine_voice —— PineClaw Voice API 的本地【模拟】客户端。
真实的 PineClaw Voice API（https://pineclaw.com/ ，作者团队开发）是一套生产级
电话语音 API：你把「电话号码 + 目标 + 上下文」交给它，它的语音 Agent 会自动完成
拨号、IVR 菜单导航（"查询请按 1，转人工请按 0"）、与真人多轮对话、实时转录，
最后把整段通话浓缩成一份【结构化通话记录】返回。
本文件不接触真实电话网络，也不需要 PineClaw key。它用 OpenAI 扮演「被叫方」
（IVR 语音菜单 + 人工客服）与「去电的语音 Agent」进行一段多轮对话，从而在本地
复现同样的输入/输出契约：
record = make_phone_call(phone_number, goal, context)
返回的 record 与真实 API 的形状保持一致（transcript / goal_achieved / key_fields ...），
因此上层 ReAct Agent 的代码在切换到真实 PineClaw SDK 时几乎无需改动。
真实接入方式见本目录 README.md「真实接入 PineClaw」一节。
"""
from __future__ import annotations
import hashlib
import json
import os
import re
import time
import uuid
from dataclasses import dataclass, field, asdict
from typing import Any
from openai import OpenAI
# 默认聊天模型：当前便宜旗舰。可用 OPENAI_MODEL 覆盖（如 Moonshot 的 kimi-k3）。
_DEFAULT_MODEL = "gpt-5.6-luna"
def _map_openrouter_model(model: str) -> str:
"""把常见模型名映射为 OpenRouter 的 provider/model 形式。"""
if "/" in model: # 已是 provider/model，原样透传
return model
if model.startswith("gpt-"):
return "openai/" + model
if model.startswith("claude-"):
return "anthropic/claude-opus-4.8"
return "openai/gpt-5.6-luna"
# 延迟创建 client：缺 Key 时由 demo.py 给出友好提示，而非 import 期裸异常。
# 通用回退：优先 OPENAI_API_KEY（可选 OPENAI_BASE_URL 指向兼容网关，如 Moonshot），
# 否则用 OPENROUTER_API_KEY 走 OpenRouter，否则给出清晰错误。
# timeout + 自动重试，避免单次网络/SSL 抖动让整段模拟通话崩溃。
_client = None
_active_model = None
def _resolve() -> tuple[OpenAI, str]:
global _client, _active_model
if _client is not None:
return _client, _active_model
model = os.getenv("OPENAI_MODEL") or _DEFAULT_MODEL
if os.getenv("OPENAI_API_KEY"):
# 直连 OpenAI（或 OPENAI_BASE_URL 指定的兼容网关，如 Moonshot kimi-k3）。
_client = OpenAI(
base_url=os.getenv("OPENAI_BASE_URL") or None,
timeout=60.0,
max_retries=3,
)
_active_model = model
elif os.getenv("OPENROUTER_API_KEY"):
# 回退到 OpenRouter：注意 gpt-5.6* 直连 OpenAI 需组织实名认证，走 OpenRouter 免此步。
_client = OpenAI(
base_url="https://openrouter.ai/api/v1",
api_key=os.getenv("OPENROUTER_API_KEY"),
timeout=60.0,
max_retries=3,
)
_active_model = _map_openrouter_model(model)
else:
raise RuntimeError(
"未检测到 OPENAI_API_KEY 或 OPENROUTER_API_KEY。请在 .env / 环境变量中至少配置其一"
"（见 env.example），或用 python demo.py --dry-run 走完全离线的脚本演示。"
)
return _client, _active_model
def _get_client() -> OpenAI:
return _resolve()[0]
def default_model() -> str:
"""当前生效的聊天模型名（已按 OpenRouter 映射解析）。"""
return _resolve()[1]
# 一次模拟通话里「去电语音 Agent ↔ 被叫方」最多来回的轮数（一轮 = 双方各说一次）。
_MAX_TURNS = 8
# 结束标记：去电语音 Agent 认为通话目标已完成/无法推进时，在发言末尾附上它。
_END_TOKEN = "[END_CALL]"
# --------------------------------------------------------------------------- #
# 结构化通话记录 —— 与真实 PineClaw Voice API 返回体保持同一形状
# --------------------------------------------------------------------------- #
@dataclass
class CallRecord:
call_id: str
phone_number: str
goal: str
status: str # "completed" / "failed"
goal_achieved: bool # 是否达成通话目标
duration_seconds: int # 通话时长（模拟值）
summary: str # 一句话通话摘要
key_fields: dict[str, Any] # 从通话中抽取的关键信息（确认号、金额、时间...）
transcript: list[dict[str, str]] # 逐轮对话：{"speaker": ..., "text": ...}
follow_up_needed: bool # 是否仍需追问/再拨
follow_up_reason: str # 若需追问，原因是什么
def to_dict(self) -> dict[str, Any]:
return asdict(self)
# --------------------------------------------------------------------------- #
# 内部：调用 OpenAI 生成一次发言
# --------------------------------------------------------------------------- #
def _chat(
messages: list[dict[str, str]],
temperature: float = 0.7,
model: str | None = None,
) -> str:
resp = _get_client().chat.completions.create(
model=model or default_model(),
messages=messages,
temperature=temperature,
)
return (resp.choices[0].message.content or "").strip()
def _callee_system_prompt(goal: str, context: str) -> str:
"""被叫方：先当自动 IVR 语音菜单，被转人工后再扮演人工客服。"""
return (
"你在参与一段【电话客服】通话的角色扮演，扮演【被叫方】（企业客服热线）。\n"
"整通电话分两个阶段，请自然衔接：\n"
"1) 开场你是【自动语音应答（IVR）系统】：先播报欢迎语，再报一个简短的按键菜单"
"（例如『账单查询请按 1，业务办理请按 2，人工服务请按 0』）。\n"
"2) 当来电者选择转人工（按 0 或明确要求人工）后，你切换为【人工客服代表】，"
"自报工号，然后正常应答。\n\n"
"扮演要求：\n"
"- 全程用简体中文口语，像真实电话一样简短自然，不要旁白、不要括号动作。\n"
"- 你可以合理编造本企业的业务细节（如扣费原因、套餐、确认号），要具体、前后一致。\n"
f"- 来电者此行的目标大致是：{goal}\n"
"- 你不必无条件满足对方，可以先解释、再看情况处理；但对方礼貌且诉求合理时，"
"应能给出明确结论（原因/处理结果/确认号等）。\n"
"- 每次只说你（被叫方）这一方的一段话，不要替对方说话。"
)
def _caller_system_prompt(phone_number: str, goal: str, context: str) -> str:
"""去电的语音 Agent：代表用户拨打电话、导航 IVR、达成目标。"""
return (
"你是 PineClaw 电话语音 Agent，正代表用户拨打一通真实电话去办事。\n"
f"- 拨打号码：{phone_number}\n"
f"- 通话目标：{goal}\n"
f"- 已知上下文：{context or '（无额外上下文）'}\n\n"
"行为要求：\n"
"- 全程用简体中文口语，礼貌、简短、直奔目标。\n"
"- 开场若遇到 IVR 按键菜单，请明确说出你的选择（如『我按 0 转人工』）来导航。\n"
"- 接通人工后清晰说明来意，围绕目标追问，主动核对并记住关键信息"
"（金额、原因、确认号、时间等）。\n"
"- 目标达成、或对方明确无法处理时，礼貌道谢结束通话，并在该句结尾附上标记 "
f"{_END_TOKEN}。\n"
"- 每次只说你（来电者）这一方的一段话，不要替对方说话，也不要附加旁白。"
)
def _run_dialog(
phone_number: str, goal: str, context: str, model: str | None = None
) -> list[dict[str, str]]:
"""驱动两个 LLM 角色来回对话，产出逐轮 transcript。"""
caller_sys = _caller_system_prompt(phone_number, goal, context)
callee_sys = _callee_system_prompt(goal, context)
# 各自维护自己视角的消息历史。
caller_msgs = [{"role": "system", "content": caller_sys}]
callee_msgs = [{"role": "system", "content": callee_sys}]
transcript: list[dict[str, str]] = []
# 被叫方先开口（IVR 欢迎语 + 菜单）。
callee_msgs.append({"role": "user", "content": "（电话已接通，请开始。）"})
callee_line = _chat(callee_msgs, model=model)
callee_msgs.append({"role": "assistant", "content": callee_line})
caller_msgs.append({"role": "user", "content": callee_line})
transcript.append({"speaker": "被叫方", "text": callee_line})
for _ in range(_MAX_TURNS):
# 去电语音 Agent 发言
caller_line = _chat(caller_msgs, model=model)
ended = _END_TOKEN in caller_line
caller_line_clean = caller_line.replace(_END_TOKEN, "").strip()
caller_msgs.append({"role": "assistant", "content": caller_line})
callee_msgs.append({"role": "user", "content": caller_line_clean})
transcript.append({"speaker": "语音Agent", "text": caller_line_clean})
if ended:
break
# 被叫方回应
callee_line = _chat(callee_msgs, model=model)
callee_msgs.append({"role": "assistant", "content": callee_line})
caller_msgs.append({"role": "user", "content": callee_line})
transcript.append({"speaker": "被叫方", "text": callee_line})
return transcript
def _extract_structured(
goal: str, transcript: list[dict[str, str]], model: str | None = None
) -> dict[str, Any]:
"""通话结束后，用一次 LLM 调用把 transcript 归纳为结构化字段。"""
dialog_text = "\n".join(f"{t['speaker']}：{t['text']}" for t in transcript)
prompt = (
"下面是一段电话通话的完整转录。请你作为通话分析器，输出一个 JSON 对象，字段如下：\n"
' "goal_achieved": 布尔，本次通话是否达成了给定目标；\n'
' "summary": 字符串，一句话中文通话摘要；\n'
' "key_fields": 对象，从通话中抽取的关键信息键值对（如 扣费原因/涉及金额/'
"确认号/处理结果/预约时间 等，键用中文，没有就留空对象）；\n"
' "follow_up_needed": 布尔，用户是否还需要进一步追问或再次拨打；\n'
' "follow_up_reason": 字符串，若需追问说明原因，否则空字符串。\n\n'
f"【通话目标】{goal}\n\n"
f"【通话转录】\n{dialog_text}\n\n"
"只输出 JSON，不要额外解释。"
)
raw = _chat(
[
{"role": "system", "content": "你是严谨的通话记录结构化器，只输出合法 JSON。"},
{"role": "user", "content": prompt},
],
temperature=0.0,
model=model,
)
structured = _safe_json(raw)
# 模型偶尔输出 JSON 数组等合法但非对象的 JSON；归一化为空 dict，
# 交给下游 .get 默认值兜底，而不是让 .get 在 list 上崩溃。
if not isinstance(structured, dict):
structured = {}
return structured
def _safe_json(raw: str) -> dict[str, Any]:
"""从模型输出里稳妥地取出 JSON（容忍 ```json 包裹）。"""
text = raw.strip()
text = re.sub(r"^```(?:json)?", "", text).strip()
text = re.sub(r"```$", "", text).strip()
try:
return json.loads(text)
except json.JSONDecodeError:
m = re.search(r"\{.*\}", text, re.DOTALL)
if m:
try:
return json.loads(m.group(0))
except json.JSONDecodeError:
pass
return {
"goal_achieved": False,
"summary": "（结构化解析失败，请查看原始 transcript）",
"key_fields": {},
"follow_up_needed": True,
"follow_up_reason": "通话记录未能被自动结构化。",
}
# --------------------------------------------------------------------------- #
# 离线 dry-run —— 完全不联网、不需要任何 API Key 的脚本化通话记录
# --------------------------------------------------------------------------- #
def _dryrun_record(phone_number: str, goal: str, context: str) -> dict[str, Any]:
"""
返回一份【脚本化】的结构化通话记录，全程不调用任何 API、不联网。
用途：当既没有电话账号、也没有 OPENAI_API_KEY 时，仍能把整条 ReAct 循环
（思考 → 调用 make_phone_call → 读取结构化记录 → 汇报）离线跑通，展示其形状。
transcript / key_fields 均为固定脚本，不代表任何真实通话。
"""
ctx = context or "（无额外上下文）"
# 由 goal 派生一个稳定、可复现的确认号（而非随机编造，便于对齐说明）。
conf = "PC" + hashlib.sha1(goal.encode("utf-8")).hexdigest()[:8].upper()
transcript = [
{"speaker": "被叫方", "text": "您好，欢迎致电客服热线。业务查询请按 1，人工服务请按 0。"},
{"speaker": "语音Agent", "text": "我按 0 转人工。"},
{"speaker": "被叫方", "text": "您好，人工客服工号 3021，请问有什么可以帮您？"},
{"speaker": "语音Agent", "text": f"你好，我这边的诉求是：{goal}。相关信息：{ctx}。"},
{"speaker": "被叫方",
"text": f"好的，已为您核实并受理，给您一个确认号 {conf}，后续可凭此查询进度。"},
{"speaker": "语音Agent", "text": f"好的，确认号 {conf}，我记下了，谢谢，再见。"},
]
record = CallRecord(
call_id="pc_dryrun_" + hashlib.sha1((phone_number + goal).encode()).hexdigest()[:8],
phone_number=phone_number,
goal=goal,
status="completed",
goal_achieved=True,
duration_seconds=12 * len(transcript) + 8,
summary=f"（dry-run 脚本模拟）已就“{goal}”联系 {phone_number} 并受理，确认号 {conf}。",
key_fields={"处理结果": "已受理", "确认号": conf, "对方号码": phone_number},
transcript=transcript,
follow_up_needed=False,
follow_up_reason="",
)
return record.to_dict()
# --------------------------------------------------------------------------- #
# 对外唯一入口 —— 对齐真实 PineClaw Voice API 的 make_phone_call
# --------------------------------------------------------------------------- #
def make_phone_call(
phone_number: str,
goal: str,
context: str = "",
*,
model: str | None = None,
dry_run: bool = False,
) -> dict[str, Any]:
"""
发起一通（模拟的）电话，由语音 Agent 全程完成，返回结构化通话记录（dict）。
参数：
phone_number: 被叫号码（本模拟不真正拨号，仅记录）。
goal: 本次通话要达成的目标，例如
"查询本月宽带账单为何多扣了 50 元并要求解释"。
context: 辅助上下文，如账号、姓名、已知信息等（可为空）。
model: 覆盖内部模拟对话所用的 OpenAI 模型（默认取 OPENAI_MODEL）。
dry_run: 若为真，则走完全离线的脚本化路径（不联网、不需要任何 API Key）。
返回：
CallRecord.to_dict()，包含 transcript / goal_achieved / key_fields 等。
"""
if dry_run:
return _dryrun_record(phone_number, goal, context)
started = time.time()
transcript = _run_dialog(phone_number, goal, context, model=model)
structured = _extract_structured(goal, transcript, model=model)
# 用对话轮数粗略折算一个「通话时长」，纯属演示展示用。
duration = 12 * len(transcript) + 8
# key_fields 可能是模型误输出的数组/标量等非对象类型，统一归一化为 dict。
key_fields = structured.get("key_fields")
if not isinstance(key_fields, dict):
key_fields = {}
record = CallRecord(
call_id=f"pc_{uuid.uuid4().hex[:12]}",
phone_number=phone_number,
goal=goal,
status="completed",
goal_achieved=bool(structured.get("goal_achieved", False)),
duration_seconds=duration,
summary=str(structured.get("summary", "")),
key_fields=key_fields,
transcript=transcript,
follow_up_needed=bool(structured.get("follow_up_needed", False)),
follow_up_reason=str(structured.get("follow_up_reason", "")),
)
_ = started # 真实 API 会用真实起止时间；此处 duration 为模拟值
return record.to_dict()
test_structured_coercion.py¶
"""回归测试：通话结构化结果类型异常时，make_phone_call 不应崩溃。
覆盖两类模型失误（此前会让 record 组装阶段 AttributeError/ValueError）：
1) 结构化输出是 JSON 数组而非对象 -> 归一化为空 dict，走默认值；
2) key_fields 是数组而非对象 -> 归一化为 {}。
不依赖真实 API：_run_dialog / _chat 被打桩。
"""
import sys
from pathlib import Path
from types import ModuleType
sys.path.insert(0, str(Path(__file__).parent))
try:
import openai # noqa: F401
except ImportError:
sys.modules["openai"] = ModuleType("openai")
sys.modules["openai"].OpenAI = object
import pine_voice
pine_voice._run_dialog = lambda *a, **k: [{"speaker": "被叫方", "text": "您好"}]
def test_json_array_structured_output_falls_back_to_defaults():
pine_voice._chat = lambda *a, **k: '[{"goal_achieved": true}]'
record = pine_voice.make_phone_call("10010", "查询账单")
assert record["goal_achieved"] is False
assert record["summary"] == ""
assert record["key_fields"] == {}
assert record["status"] == "completed"
def test_non_dict_key_fields_coerced_to_empty_dict():
pine_voice._chat = lambda *a, **k: (
'{"goal_achieved": true, "summary": "s", "key_fields": ["确认号", "金额"],'
' "follow_up_needed": false, "follow_up_reason": ""}')
record = pine_voice.make_phone_call("10010", "查询账单")
assert record["goal_achieved"] is True
assert record["key_fields"] == {}
def test_wellformed_structured_output_unchanged():
pine_voice._chat = lambda *a, **k: (
'{"goal_achieved": true, "summary": "已受理", '
'"key_fields": {"确认号": "PC123"}, '
'"follow_up_needed": false, "follow_up_reason": ""}')
record = pine_voice.make_phone_call("10010", "查询账单")
assert record["goal_achieved"] is True
assert record["summary"] == "已受理"
assert record["key_fields"] == {"确认号": "PC123"}