企业官网建设流程全解析

Vision Agents 实战：用 Python 构建静默监听式 AI 会议教练（Sales Assistant 示例全解析）

【免费下载链接】Vision-AgentsOpen Vision Agents by Stream. Build voice and vision agents quickly with any model or video provider. Uses Stream's edge network for ultra-low latency.项目地址: https://gitcode.com/GitHub_Trending/vi/Vision-Agents

本篇技术指南以仓库中 examples/09_sales_assistant_example 的完整示例为主体，讲解如何基于 Vision Agents 构建一个"AI Meeting Copilot"：它能在会议、面试和销售通话中静默监听麦克风与系统音频，通过 AssemblyAI 完成带说话人分离的实时转写，再由 Gemini 生成"接下来该说什么"的教练建议，并同步到 Stream Chat 显示在半透明 macOS 覆盖层上。读完本文，你将掌握一个非实时 STT + LLM 推理链路的完整搭建方法、Agent HTTP 服务器与会话接口的设计方式，以及如何通过系统提示词（instructions.md）约束 AI 教练的输出风格。

示例是什么：AI 会议教练（Sales Assistant）

09_sales_assistant_example是一个实时 AI 副驾（real-time AI copilot）示例：在会议、面试、销售通话进行期间，它静默监听你的麦克风和系统音频（对其他参会者不可见），把对话转写成文本并做说话人分离（speaker diarization），随后分析对话内容，在一个半透明 macOS 覆盖层（overlay）上实时给出教练建议——例如"现在该说什么""如何过渡到下一个话题"。

该 agent 可以被扩展以接入 RAG 与自定义知识库，从而让建议贴合你自己的产品、公司打法手册（playbook）或具体交易上下文。也就是说，它从一个通用教练变成了懂你业务的专属顾问。

说明：本示例只涉及Python Agent这一侧（即本目录）；配套的 macOS 覆盖层应用位于独立的 companion 演示仓库中，README 中给出了其地址，本文按当前仓库约束只展开 Python Agent 侧的实现。

系统架构：Python Agent + macOS 覆盖层

项目由两个组件组成，分工明确：

整体运行流程（5 步）：

1. 用户打开 macOS 覆盖层，点击Start；
2. 应用创建一个带屏幕共享（包含系统音频捕获）的 Stream Video 通话；
3. 应用告知 Python Agent 服务器加入该通话；
4. Agent 转写音频（AssemblyAI STT + diarization）并生成教练建议（Gemini LLM）；
5. 教练建议以文本形式通过Stream Chat显示在半透明覆盖层上。

这种"端侧捕获音频 + 云端 Agent 分析"的拆分，使得 AI 的加入对通话中的其他参与者完全不可见——他们看不到覆盖层，也听不到任何 TTS 声音，只会感觉到你"突然说得很到位"。

核心 AI 流水线：非实时 STT + LLM

该示例的关键设计在于使用了一条非实时（non-realtime）STT + LLM 流水线，与实时语音对话流水线相比更加轻量：

• AssemblyAI STT：把会议音频转写为文本，并启用说话人分离（speaker_labels=True）。在 AssemblyAI 插件文档 中可以看到，启用后每个转录事件会为不同说话人携带独立的participant，原始标签存放在response.other["speaker_label"]中；
• Gemini LLM：分析转录文本，生成 1–3 句的简短教练建议（示例使用gemini-flash-lite-latest模型，见 main.py）；
• Stream Chat 同步：LLM 响应由 Vision Agents SDK 自动同步到messaging:{callId}频道（见 stream_edge_transport.py 中self.client.chat.channel("messaging", call.id)的实现），macOS 覆盖层监听该频道并渲染文本；
• 无需 TTS：因为建议以文本显示而非语音播放，整个流水线不需要文本转语音，也让 WebRTC 配置保持简单——Agent 只需一个订阅（subscriber）连接，无需发布（publisher）音频轨道。

提示（来自官方 README）：如果希望增加屏幕分析能力，可以把gemini.LLM换成gemini.Realtime(fps=3)。需要注意：Realtime 模式会额外输出音频，Agent 除了把建议写入聊天外，还会把建议朗读出来。

环境与前置条件

运行本示例需要满足以下条件：

• macOS 13.0 或更高版本（因为覆盖层应用是 macOS 原生应用）；
• Python 3.12+，推荐使用 uv（uv 是 Astral 出品的包管理器，本示例的pyproject.toml中requires-python = ">=3.12"）或 pip；
• 三组 API Key：
  • Stream：Video API key + secret（用于通话与 Chat 同步）；
  • Google AI Studio：Gemini API key（用于 LLM 分析）；
  • AssemblyAI：STT API key（用于转写）。

环境变量清单

从 main.py 与插件源码可以确认以下环境变量（按需填入.env文件）：

事实提示：README 中列出项目结构包含.env.example模板，但当前仓库该目录实际只包含README.md、instructions.md、main.py、pyproject.toml四个文件，并未随附.env.example。如需模板可参考仓库内其他示例（如 examples/04_football_commentator_example/env.example 和 examples/11_moderation_example/env.example），或直接按上表变量名自行创建.env文件。

快速开始：从零启动 coaching server

1. Python Agent（本仓库）

# 复制并填写 API Key（如仓库无该模板文件，请按上文环境变量表手动创建） cp .env.example .env # 用编辑器编辑 .env，填入 Stream / Gemini / AssemblyAI 的 key # 安装依赖（使用 uv） uv sync # 启动 Agent HTTP 服务器 uv run main.py serve

服务器默认监听http://localhost:8000。macOS 应用会调用POST /sessions来启动教练会话（README 中的简化表述；当前 SDK 的完整路由为POST /calls/{call_id}/sessions，详见下文"HTTP 接口"一节）。

依赖清单定义在 pyproject.toml 中：vision-agents核心库 +vision-agents-plugins-assemblyai、vision-agents-plugins-getstream、vision-agents-plugins-gemini三个插件，以及python-dotenv（加载.env）和getstream（SDK 客户端）。值得注意的是，[tool.uv.sources]中将三个依赖都指向了仓库内的../../agents-core与../../plugins/*并以editable = true方式安装，即直接使用本仓库的源码运行。

2. macOS 覆盖层应用

配套的 macOS 应用位于独立的 companion 演示仓库中，需要按该仓库 README 中的构建与运行说明来执行。该应用预期 Agent 服务器运行在http://localhost:8000，所以请先启动 Python Agent。

serve 命令的可用参数

serve子命令由 Runner CLI 提供（见 runner.py），支持以下选项：

使用指南：一次完整的教练会话

1. 启动 Python Agent 服务器（终端 1）：

   uv run main.py serve

2. 运行 macOS 覆盖层——参见 companion 应用仓库的说明（其 README 中描述了构建与运行步骤）；
3. 屏幕右上角出现半透明覆盖层窗口；
4. 点击Start开始一次教练会话，应用将自动：
   • 共享你的屏幕（包含系统音频）；
   • 连接 AI Agent；
   • 实时显示陆续到达的教练建议；
5. 点击Stop结束会话。

HTTP 接口：会话生命周期

当uv run main.py serve启动后，Agent 通过 Runner 的 HTTP 服务暴露会话接口。当前 SDK 的核心路由定义在 api.py：

• POST /calls/{call_id}/sessions：启动新 Agent 并让其加入指定通话（对应 README 中"POST /sessions"的完整形态），成功返回201与session_id、call_id、session_started_at；超出并发或单通话会话上限时返回429；
• DELETE /calls/{call_id}/sessions/{session_id}（以及同路径的 POST.../close，兼容浏览器 sendBeacon 场景）：请求关闭会话，返回202；
• GET /calls/{call_id}/sessions/{session_id}：查询运行中会话的信息；
• GET /calls/{call_id}/sessions/{session_id}/metrics：从注册表（registry）读取会话指标；
• GET /health/GET /ready：存活与就绪探针（/ready在 launcher 尚未就绪时返回503）。

这些接口的鉴权行为由ServeOptions控制（options.py），包括can_start_session、can_close_session、can_view_session、can_view_metrics四个可调用对象（默认放行），以及 CORS 配置（默认允许所有来源与方法）。

此外，本示例还通过runner.fast_api挂载了两个自定义端点（见下文源码剖析），用于注入会议上下文与签发客户端 token。

源码剖析：main.py 逐段解读

main.py 是示例的全部后端代码，只有约 110 行，结构非常清晰。

1. Agent 定义：create_agent

async def create_agent(**kwargs) -> Agent: agent = Agent( edge=getstream_edge.Edge(), agent_user=User(name="Sales Assistant", id="sales-assistant-agent"), instructions="Read @instructions.md", llm=gemini.LLM("gemini-flash-lite-latest"), stt=assemblyai.STT(speaker_labels=True), ) return agent

• edge使用 getstream 插件的Edge()传输层，负责 Stream 边缘网络上的音视频连通；
• agent_user定义了 Agent 在 Stream 中的身份；
• instructions="Read @instructions.md"：使用@指令语法引用 Markdown 文件作为系统提示词。SDK 的 instructions.py 中用正则@([^\s@]+)解析所有@提及的.md文件，并把文件内容合并进完整提示词（忽略以.开头的文件、非 md 文件以及 base 目录外的文件）；
• llm使用 Gemini 的 flash-lite 轻量模型；
• stt=assemblyai.STT(speaker_labels=True)开启说话人分离。AssemblyAI 插件还支持max_speakers（1–10 的说话人数量提示，需同时开启speaker_labels）、speech_model（默认u3-rt-pro）、sample_rate（默认 16000 Hz）等参数，插件测试 test_assemblyai_stt.py 验证了 speaker labels 的默认关闭与开启行为。

注释中特别说明：不需要 TTS——Agent 把建议写入 Stream Chat 而非发声，这也让 WebRTC 配置保持简单（只有 subscriber 连接，没有 publisher 音频轨道）。

2. 加入通话：join_call

async def join_call(agent, call_type, call_id, **kwargs) -> None: call = await agent.create_call(call_type, call_id) async with agent.join(call): prompt = ( "Listen to the conversation. " "Provide real-time coaching suggestions — tell the user what to say next. " "Keep every suggestion to 1-3 sentences." ) if _meeting_context: prompt = f"Meeting context: {_meeting_context}\n\n{prompt}" await agent.simple_response(prompt) await agent.finish()

• agent.create_call(call_type, call_id)创建通话对象；
• agent.join(call)以上下文管理器方式加入通话，进入后即开始教练；
• 先拼接一段启动指令：倾听对话、给出实时教练建议、告诉用户接下来说什么、每条建议保持 1–3 句；如果 macOS 应用预先设置了会议上下文（_meeting_context），则作为前缀注入 prompt，让建议贴合当前会议背景；
• agent.simple_response(prompt)把注入指令交给 LLM。SDK 中 Agent.simple_response 会把请求路由到推理流（inference flow），与语音驱动的回合共享同一套 LLM/TTS/音频管线；participant缺省时归属 Agent 自己，interrupt=True时会抢占进行中的 LLM 回合。LLM 的响应会自动同步到messaging:{call_id}频道；
• agent.finish()保持 Agent 存活等待通话结束（agents.py），此后 STT + 轮次检测（turn detection）会自动触发额外的simple_response调用，实现持续的实时建议。

3. Runner 与自定义端点

runner = Runner(AgentLauncher(create_agent=create_agent, join_call=join_call))

Runner（runner.py）负责以 HTTP 服务器模式运行 Agent，AgentLauncher（agent_launcher.py）负责预热与生命周期管理——确保 LLM、TTS、STT、轮次检测等组件在接收会话前完成初始化，这也是/ready探针返回 503 与否的依据。

示例额外暴露了两个端点，供 macOS 应用使用：

@runner.fast_api.put("/context") async def set_context(request: dict) -> JSONResponse: global _meeting_context _meeting_context = request.get("context", "") return JSONResponse({"ok": True}) @runner.fast_api.get("/auth/token") async def create_token(user_id: str = Query(...)) -> JSONResponse: token = _stream_client.create_token(user_id) return JSONResponse({"token": token, "apiKey": _api_key})

• PUT /context：在会话开始前设置"会议上下文"（如客户公司、谈判要点、本次面试职位），上下文会注入到join_call的启动 prompt 中；
• GET /auth/token?user_id=...：为 Flutter 客户端生成 Stream 用户 token，并返回apiKey，使客户端能使用与 Agent 相同的 Stream 应用获取凭证。

最后runner.cli()把 CLI 入口（serve等子命令）绑定到脚本，使得uv run main.py serve可直接启动服务器。

提示词设计：instructions.md 系统提示

instructions.md 是整个教练"人设"的灵魂，也是被Read @instructions.md注入的完整系统提示词：

角色定义：你是 Sales Assistant——一个实时会议与面试教练。你监听用户在实时会议、面试或通话中的麦克风，任务是在对话进行中给出简短、可执行的教练建议——告诉用户"接下来该说什么"。

六条行为规则：

1. 保持简洁：每条建议必须 1–3 句话。用户是在对话进行中实时阅读你的建议，读不了段落；
2. 聚焦"说什么"：不要描述你听到了什么，也不要复述会议内容——直接给用户能说出口的话；
3. 主动出击：发现有人提问就建议答案；察觉犹豫就建议自信的回应；冷场时就建议一个过渡话题；
4. 利用转录：音频转录告诉你已说过的话，用它保持上下文、避免重复已谈过的要点；
5. 保持隐身：永远不要提到你自己、覆盖层或教练系统，把建议写成用户自己的内心声音；
6. 适应场景：无论是求职面试、销售电话、团队站立会还是一对一沟通，都要调整教练风格以匹配情境。

四条示例建议（展示期望的输出形态）：

• "Say: 'That's a great point — we saw a 20% improvement after implementing that change.'"（直接给出可说的话术）
• "Mention your experience with distributed systems here."（提示应提及的经历）
• "Ask them: 'What does success look like for this role in the first 90 days?'"（建议向对方提出的问题）
• "Wrap up this point and transition to the pricing discussion."（引导话题过渡）

这套提示词的设计要点在于：输出必须是"可直接朗读的短句"，且绝不能让对方察觉 AI 的存在——这与"半透明覆盖层 + 无 TTS"的产品形态是自洽的。

扩展方向

• RAG 与自定义知识库：官方 README 明确指出 agent 可以扩展 RAG 与自定义知识库，让建议贴合你的产品、公司打法手册或具体交易上下文。仓库内提供了多种 RAG 插件（如 qdrant、turbopuffer）可参考；
• 屏幕分析：把gemini.LLM换成gemini.Realtime(fps=3)即可让 agent 看到共享屏幕画面（例如演示中的演示文稿），实现基于画面的实时建议；注意 Realtime 模式会同时输出音频，agent 会把建议朗读出来；
• 替换模型与 STT 提供商：Vision Agents 提供了丰富的插件生态（openai、anthropic、deepgram、elevenlabs 等 30+ 插件，均在 plugins 目录下），可以按预算、时延和语言需求自由组合 LLM 与 STT。

项目结构速览

examples/09_sales_assistant_example/ ├── main.py # Agent 定义 + HTTP 服务器（create_agent / join_call / Runner） ├── instructions.md # 教练 Agent 的系统提示词（@ 指令引用的 Markdown） ├── pyproject.toml # 依赖清单（uv 管理，指向仓库内源码） └── README.md # 本示例的完整使用文档

（README 中列出的.env.example模板在当前仓库未随附，请按上文"环境变量清单"自行创建，或参考 examples/04_football_commentator_example/env.example 的格式。）

小结

09_sales_assistant_example展示了 Vision Agents 在"语音转写 + 分析 + 文本回传"这一非实时推理链路上的典型用法：AssemblyAI 负责带说话人分离的转写，Gemini 负责基于转录的实时教练建议，Stream 边缘网络负责通话与 Chat 同步，main.py以约百行代码将其串成完整的 HTTP 服务。配合精心设计的系统提示词，一个对其他参会者完全不可见的 AI 会议教练就此诞生——这也是将 Vision Agents 用于会议副驾、面试陪练、销售陪练等场景的绝佳起点。

【免费下载链接】Vision-AgentsOpen Vision Agents by Stream. Build voice and vision agents quickly with any model or video provider. Uses Stream's edge network for ultra-low latency.项目地址: https://gitcode.com/GitHub_Trending/vi/Vision-Agents