用 Python 为 AI 电话任务建立审计账本
开源示例 missions-audit 基于 Python 和 Telnyx AI Missions,记录任务目标、执行计划、状态变化及类型化事件,并生成带时间戳的审计报告。其核心是将审计记录纳入工作流 API 契约。
完整中文译文
AI 电话任务返回成功,不等于你能证明电话打给了谁、执行了哪些步骤、最后观察到了什么。Telnyx 的开源 missions-audit 示例,把目标、执行计划、步骤状态、事件和关联助手存成可查询的记录,再生成审计报告。
它最值得借鉴的地方,是把审计记录纳入工作流的 API 契约。对于需要追查 AI 自动化执行过程的团队,这比多打印几行日志更有参考价值。不过,这套记录能证明什么,仍要看执行器实际观察到了什么。
一次成功响应,为什么还需要一套执行记录
原文用诊所预约外呼举例:助手联系患者、确认预约,然后报告任务完成。审核者需要追问:原定目标是什么,计划包含哪些步骤,执行时发生了哪些状态变化,哪个助手参与了任务?
应用日志可能包含这些信息,但通常散落在不同请求和进程中。missions-audit 采用了更明确的结构:
- Mission 保存长期存在的任务目标。
- Run 表示对该目标的一次执行。
- Plan 保存有序步骤及其状态。
- Events 保存带类型的执行事件。
- Linked agents 记录本次执行关联的助手。
大白话解释,就是把「要做什么」「这次怎么做」「做到了哪一步」「观察到什么」分别存好,并用标识关联起来。
示例使用 Python,通过原始 HTTP 请求访问 Telnyx /v2 API。创建目标用 POST /ai/missions,其中 execution_mode 设为 "external":由应用负责执行动作,Missions API 负责保存持久记录。
创建一次执行用 POST /ai/missions/{mission_id}/runs。Run 从 pending 开始,可进入 running,最终成为 succeeded 或 failed;API 还支持 paused 和 cancelled,供人工介入。
计划通过 POST /ai/missions/{mission_id}/runs/{run_id}/plan 提交。客户端指定 step_id、描述和从 1 开始的 sequence。示例分成三步:provision_assistant、schedule_outreach_call、verify_outcome,对应准备助手、安排电话、验证结果。
步骤也有独立状态:从 pending 到 in_progress,再到 completed 或 failed;有意不执行的步骤可以标为 skipped。任务状态与步骤状态分开后,报告才能回答「整体失败具体卡在哪里」。
最关键的约束,是执行器不能替平台宣布成功
原文最有价值的规则是:记录执行器观察到的变化,不编造平台结果。
下面这段代码把工作函数包起来,让步骤开始、异常和完成都有对应的状态与事件:
def run_step(step_id, work):
update_step(step_id, "in_progress")
log_event("step_started", f"Step {step_id} started", step_id)
try:
result = work()
except Exception as exc:
update_step(step_id, "failed")
log_event(
"step_failed",
f"Step {step_id} failed: {exc}",
step_id,
)
raise
update_step(step_id, "completed")
log_event("step_completed", f"Step {step_id} completed", step_id)
return result
要留意 result = work():包装器只能根据这个函数是否抛出异常判断步骤结果。work() 内部必须认真验证业务结果,否则一个提前返回的函数,仍会让步骤被标成完成。
真实外呼模式下,执行器每 5 秒轮询一次 scheduled-event API,最多等待 10 分钟。每次观察到新的事件状态,就追加一个 status_change 事件。请求失败或超时会产生明确的 error 事件,并让验证步骤失败。
没有观察到通话完成,就不能因为预期它会完成而记成完成。
不过,轮询记录的是「被观察到的状态」,未必包含两次查询之间的每次平台变化。原文也没有说明各类电话终态如何对应「联系到患者」「确认了预约」这些业务结论。事件进入终态,与业务目标真正达成之间,还需要明确的判断规则。
演示完成和真实通话完成,必须在记录里分得清
项目默认 DEMO_MODE=true。此时会调用 mission、run、plan、step 和 event 相关 API,但不会创建助手、创建 scheduled event、拨打电话或关联 agent。
演示会写入 custom 事件,包含 "demo_mode": true,以及 "note": "no assistant created, no call placed"。Run 的 result_summary 也会明确说明没有真实外呼。
这个设计解决了一个实际问题:演示流程可以跑通,但不能把演示记录当作电话执行证据。它仍会创建远端任务记录,所以这里的演示边界主要是没有真实电话动作。
切换真实路径需要设置 DEMO_MODE=false,并配置 TELNYX_API_KEY=YOUR_TELNYX_API_KEY、DESTINATION_NUMBER=<E164_TEST_NUMBER> 和 ASSISTANT_PHONE_NUMBER=<E164_ASSISTANT_NUMBER>。原文要求使用自己控制的测试号码,并提前配置助手号码。
真实执行器可以创建或复用助手、安排电话事件、轮询状态、记录最近一次通话尝试,并将助手关联到 Run。接接口时还有个容易踩坑的差异:Missions 响应使用 data 包装,标识包括 mission_id、run_id;Assistants 和 scheduled-events 返回直接的 JSON 对象,需要分别解析。
项目将流程拆成五个脚本。进入 telnyx-code-examples/missions-audit 后,先用 python -m venv .venv 创建环境,执行 source .venv/bin/activate、pip install -r requirements.txt 和 cp .env.example .env,填写 API key,再运行 pytest smoke_test.py -q。
下面这些命令串起完整流程,后一步使用前一步输出的标识:
python create_mission.py
python start_run.py --mission-id <MISSION_ID>
python plan_steps.py --mission-id <MISSION_ID> --run-id <RUN_ID>
python execute.py --mission-id <MISSION_ID> --run-id <RUN_ID>
python audit.py --mission-id <MISSION_ID> --run-id <RUN_ID>
要留意每一步的 --mission-id 和 --run-id:它们决定计划、执行事件和报告是否归属于同一次任务。
报告为什么要重新读取服务端状态
audit.py 不直接拿执行器内存里的结果拼报告。它先等待 Run 进入 succeeded、failed 或 cancelled,再重新读取 mission、run、plan、events 和关联 agents,生成带时间戳的审计报告。
报告包含任务标识、Run 状态、result_summary、各步骤结果、事件时间线,以及 run_id 对应的 telnyx_agent_id。审核者拿到的是服务端当前保存的视图,而不只是某个进程结束前的自述。
它同样设置了 10 分钟等待上限。如果到时仍未进入终态,会打印警告、最后刷新一次状态、展示当前记录,并以非成功退出状态结束。
这种行为值得保留:等待结束只是报告程序停止等待,不能据此宣布任务完成。
我的判断
我认同作者把目标、计划、事件和助手关联建成独立资源的思路。相较于单靠应用日志排查,它提供了稳定的查询入口,也更方便审核一次执行的全貌。原文列出的预约提醒、催收联系、事故响应、销售流程和运维自动化,都有类似需求。
但原文展示的是示例设计和操作步骤,没有给出与普通日志、链路追踪或工作流引擎的对照数据,也没有公布真实外呼的测试结果。它能说明一种实现方式,不能据此认定成本更低、可靠性更高,或已经满足某类合规要求。
代码还有一个生产环境必须补上的问题:状态更新和事件写入是两次调用。如果步骤状态更新成功,随后 log_event 失败,记录就可能出现缺口。原文没交代幂等、重试、并发更新和补偿策略。
另外,开头提到审核者可能需要「已批准的计划」,示例只展示了提交计划,没有展示审批流程。可查询的记录,也不自动等于不可篡改的证据。接入已有工作流引擎的团队,可以先评估现有执行历史是否已覆盖这些需求,再决定是否增加一套任务资源。
能拿来做什么
- 做 AI 外呼原型时,先用
DEMO_MODE=true验证任务、步骤和报告链路,并把演示标识写进最终结果,避免测试记录被当作真实通话凭证。 - 用 Node 实现类似执行器时,可以沿用
run_step的包装思路,同时处理状态更新与事件写入之间的失败,确保重试不会重复制造业务动作。 - 给运维或事故响应 Agent 增加报告入口时,应重新读取持久状态,并展示失败步骤、超时和关联助手,方便接手的人继续排查。
- 在日常 AI 编程中,可以让助手参照这套结构生成执行器,但要单独审查
verify_outcome的判断条件,明确平台状态能证明哪些业务结果。