去年给一家电商企业做售后客服 Agent 时我们踩过一个特别典型的坑线上纯文本客服上线三个月用户满意度只有 32%。翻后台日志发现超过六成的用户咨询会附带故障截图、语音描述而客服只能回一句“抱歉我暂时无法识别图片/语音”。团队当时硬编码了 OCR 和 Whisper把识别结果拼进文本再丢给大模型结果用户同时发三张截图加一段语音时OCR 内容乱序、Whisper 对方言识别率只有 40%Agent 的回答完全答非所问。正是在这个项目里我们把多模态输入收拢到 AI Agent Harness Engineering 的架构里同时把多模态大模型调用都接到 TaoTokenhttps://taotoken.net/?utm_sourcetaotoken_aicg_blog_end这一个兼容通道上才把故障图片、语音和文本真正对齐成同一段上下文。1. 硬编码 OCR/Whisper 为什么把线束拧成了乱麻多模态售后客服最容易被低估的是“输入顺序”和“上下文归属”。用户发来一段语音说“电饭煲煮饭跳到保温后米饭还是夹生”同时拍了一张面板按钮的故障截图这两条信息在时间上属于同一次咨询。硬编码方案的处理方式是把图片丢给 OCR、语音丢给 Whisper再把两个结果一前一后拼到文本里。问题是 OCR 和 Whisper 的耗时不一样谁先返回谁就排在前面用户的语音提到“就是这个按钮”但截图内容还在异步处理中大模型看到的是语义断裂的两段文字。更麻烦的是记忆割裂。同一用户上周发过一张电源线烧焦的图片这周说“上次那个问题又出现了”如果文本记忆和图像记忆分开存储检索时只能命中文字记录找不到对应图片。我们最初的代码里图像描述、语音转写、文本提问分别散在 10 多个文件中想加一个视频输入能力改动会波及 Agent 核心推理逻辑。Harness 架构做的事情是把这些散落的模态处理逻辑收拢到统一接入层就像汽车整车线束把灯光、车窗、传感器的线路归拢到同一束线缆里。核心推理层不再关心来的是图片还是语音只面向统一格式的融合数据。这个方案有三层融合输入层把图像和语音统一转成文本表示特征层把不同模态映射到同一向量空间输出层根据用户偏好把回复转成文本、示意图或语音。我们这次重构的主线就是按这个架构把五个组件搭起来并让它们共用同一个 OpenAI 兼容通道。2. 准备材料环境依赖 一把 TaoToken Key2.1 环境依赖清单原文项目锁定了一个可复现的版本组合我们这次沿用同样组合避免后续排查问题时出现“我机器上好好的”这种对话依赖版本要求用途Python3.10.x ~ 3.11.x运行环境LangChain0.2.10Agent 核心开发框架OpenAI SDK1.35.13调用 GPT-4o 多模态、TTS 与 embeddingsTransformers4.42.4加载 CLIP 图像特征模型FAISS-CPU1.8.0多模态向量检索FastAPI0.111.0对外提供 HTTP 接口FFmpeg6.0语音/视频格式转换安装命令如下pip install langchain0.2.10 openai1.35.13 transformers4.42.4 faiss-cpu1.8.0 fastapi0.111.0 uvicorn0.30.1 python-multipart0.0.9 # Ubuntu 安装 FFmpeg其他系统请换成对应包管理命令 sudo apt update sudo apt install ffmpeg -y2.2 从 TaoToken 创建 API Key准备工作的最后一步是配置模型访问。原文那一步是export OPENAI_API_KEY你的OpenAI API Key现在改成去 TaoToken 注册并创建一个 API Key。之所以把 Key 统一走 TaoToken是因为这套 Harness 里有三处地方要调大模型InputAdapter 里的图片描述调用 GPT-4o chat、ToolOrchestration 里的 OCR 与文本回复继续走 GPT-4o chat另外 TTS 语音生成和文本 embedding 也要调 OpenAI 接口。如果分别配三把 Key、三个 Base URL排障时很难定位是哪一路出了问题。走 TaoToken 之后所有 OpenAI SDK 调用共用同一个 client 实例相当于把电源、信号、接地三条线收到同一束线缆里代码里只出现一套地址和一把 Key。模型 ID 不要从旧文档里猜以 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_end 的模型广场为准那里会列出当前可用的 GPT-4o 多模态、TTS、embedding 模型标识。2.3 初始化 OpenAI client 指向 TaoToken不管代码之前在多少个文件里创建过 OpenAI client现在统一改成这个写法from openai import OpenAI client OpenAI( api_keyYOUR_API_KEY, # 从 TaoToken 官网创建 base_urlhttps://taotoken.net/api # 注意末尾不要加 /v1 )这里的关键点base_url是填进 SDK 的接口地址不是浏览器访问的官网地址。官网https://taotoken.net/?utm_sourcetaotoken_aicg_blog_end用来注册、创建 Key、看模型广场和用量SDK 里填https://taotoken.net/api。两者不能混用更不能在api后面再加/v1否则 OpenAI SDK 拼路径时会出现双/v1导致 404。3. 五组件代码从 InputAdapter 到 ToolOrchestration3.1 InputAdapter三种模态进统一特征出InputAdapter 的职责是把文本、图片、语音转换成统一结构既保留content给大模型直接读也提取feature供多模态记忆检索。图片部分保留 CLIP 提取特征同时用 GPT-4o 生成故障描述语音部分用 Whisper 转写文本并取 encoder 特征向量。import torch import whisper from PIL import Image from transformers import CLIPProcessor, CLIPModel whisper_model whisper.load_model(base) clip_model CLIPModel.from_pretrained(openai/clip-vit-base-patch32) clip_processor CLIPProcessor.from_pretrained(openai/clip-vit-base-patch32) device cuda if torch.cuda.is_available() else cpu clip_model.to(device) # 统一特征空间维度实际项目中可微调 W_t torch.randn(2048, 3072, devicedevice) W_i torch.randn(2048, 512, devicedevice) W_s torch.randn(2048, 1280, devicedevice) b torch.randn(2048, devicedevice) class InputAdapter: async def process_text(self, text: str): resp client.embeddings.create(inputtext, modelYOUR_EMBEDDING_MODEL) vec torch.tensor(resp.data[0].embedding, devicedevice) feature torch.matmul(W_t, vec) b return {modal: text, content: text, feature: feature.cpu().numpy()} async def process_image(self, image_path: str): image Image.open(image_path) inputs clip_processor(imagesimage, return_tensorspt).to(device) image_vec clip_model.get_image_features(**inputs).squeeze() feature torch.matmul(W_i, image_vec) b response client.chat.completions.create( modelYOUR_MODEL_ID, # 以 TaoToken 模型广场为准 messages[{ role: user, content: [ {type: text, text: 100字以内描述这张图片重点关注商品故障信息}, {type: image_url, image_url: {url: ffile://{image_path}}} ] }], max_tokens200 ) return {modal: image, content: response.choices[0].message.content, feature: feature.cpu().numpy()} async def process_audio(self, audio_path: str): result whisper_model.transcribe(audio_path) audio_tensor torch.tensor(result[segments][0][audio]).unsqueeze(0).to(device) audio_vec whisper_model.encoder(audio_tensor).squeeze().mean(dim0) feature torch.matmul(W_s, audio_vec) b return {modal: audio, content: result[text], feature: feature.cpu().numpy()}这段代码里只有 Whisper 的本地转写不经过 TaoToken其余 GPT-4o 图像描述、文本 embedding 都走统一 client。3.2 ModalScheduler优先级与时间戳对齐用户同时发文本、截图、语音时不能按处理完成顺序拼 prompt而要先按时间戳分组再按优先级排序。分组时间窗我们按 10 秒误差处理语音优先级高于图片、高于文本。class ModalScheduler: def __init__(self, w10.3, w20.5, w30.2): self.w1, self.w2, self.w3 w1, w2, w3 self.modal_priority {audio: 8, image: 7, text: 5} def calculate_priority(self, user_level: int, modal_type: str, urgency: int 1): return (self.w1 * user_level self.w2 * self.modal_priority.get(modal_type, 5) self.w3 * urgency) def align_inputs(self, inputs: list): groups {} for item in inputs: ts item[timestamp] // 10000 groups.setdefault(ts, []).append(item) aligned [] for ts in sorted(groups.keys()): group groups[ts] group.sort(keylambda x: x[priority], reverseTrue) aligned.append(group) return aligned对齐之后同一次咨询里的语音和截图在 prompt 中是相邻的大模型能正确理解“就是这个按钮”指的哪张图。3.3 MultimodalMemory 与 ToolOrchestration多模态记忆模块把每次交互的特征向量和元数据存进 FAISS检索时同时召回文本、图片、语音历史。工具编排层负责按需调用 OCR、TTS、图片生成并支持熔断降级。import faiss import numpy as np class MultimodalMemory: def __init__(self, dimension2048): self.index faiss.IndexFlatL2(dimension) self.metadata [] def add(self, feature: np.ndarray, metadata: dict): self.index.add(feature.reshape(1, -1)) self.metadata.append(metadata) def retrieve(self, query_feature: np.ndarray, top_k5): if self.index.ntotal 0: return [] _, indices self.index.search(query_feature.reshape(1, -1), top_k) return [self.metadata[i] for i in indices[0] if i len(self.metadata)] class ToolOrchestration: def __init__(self): self.tools {ocr: self.call_ocr, tts: self.call_tts} async def call_ocr(self, image_path: str): response client.chat.completions.create( modelYOUR_MODEL_ID, messages[{ role: user, content: [ {type: text, text: 识别这张图片中的所有文字返回纯文本}, {type: image_url, image_url: {url: ffile://{image_path}}} ] }], max_tokens500 ) return response.choices[0].message.content async def call_tts(self, text: str, output_path: str): response client.audio.speech.create( modelYOUR_TTS_MODEL, voicealloy, inputtext ) response.stream_to_file(output_path) return output_path async def call_tool(self, tool_name: str, **kwargs): try: return await self.tools[tool_name](**kwargs) except Exception as e: print(fTool {tool_name} failed: {e}, fallback to LLM) return None这里所有 OpenAI 调用都在同一个client上所以 Key 统一走 TaoToken 之后OCR、TTS、对话生成在连接层面没有第二套地址需要维护。3.4 FastAPI 主服务与 Swagger 入口主服务接收多模态输入异步处理后把结果存到任务字典。这里沿用原文的异步处理思路但把 client 初始化替换成 TaoToken 配置并把模型 ID 参数化。from fastapi import FastAPI, File, UploadFile, Form import uuid, os, asyncio app FastAPI(title多模态售后客服 Agent) input_adapter InputAdapter() scheduler ModalScheduler() memory MultimodalMemory() tools ToolOrchestration() tasks {} app.post(/api/v1/multimodal/input) async def multimodal_input( session_id: str Form(...), user_id: str Form(...), text: str Form(None), image: UploadFile File(None), audio: UploadFile File(None), timestamp: int Form(...), output_preference: str Form(text) ): task_id str(uuid.uuid4()) tasks[task_id] {status: processing, result: None} asyncio.create_task(process_task(task_id, session_id, user_id, text, image, audio, timestamp, output_preference)) return {code: 200, message: 请求已接收, task_id: task_id} async def process_task(task_id, session_id, user_id, text, image, audio, timestamp, output_preference): try: inputs [] if text: data await input_adapter.process_text(text) inputs.append({data: data, priority: scheduler.calculate_priority(5, text), timestamp: timestamp}) if image: path f./tmp/{uuid.uuid4()}_{image.filename} with open(path, wb) as f: f.write(await image.read()) data await input_adapter.process_image(path) inputs.append({data: data, priority: scheduler.calculate_priority(5, image), timestamp: timestamp}) if audio: path f./tmp/{uuid.uuid4()}_{audio.filename} with open(path, wb) as f: f.write(await audio.read()) data await input_adapter.process_audio(path) inputs.append({data: data, priority: scheduler.calculate_priority(5, audio), timestamp: timestamp}) aligned scheduler.align_inputs(inputs) if inputs: avg_feature np.mean([x[data][feature] for x in inputs], axis0) history memory.retrieve(avg_feature, top_k3) else: history [] prompt 你是专业的电商售后客服。结合历史上下文和当前输入给出解决方案。\n for item in history: prompt f- 历史{item[content]}\n for group in aligned: for item in group: prompt f- {item[data][modal]}{item[data][content]}\n response client.chat.completions.create( modelYOUR_MODEL_ID, messages[{role: user, content: prompt}], max_tokens1000 ) reply response.choices[0].message.content result {text: reply, image_url: None, audio_url: None} if output_preference audio: audio_path f./output/{uuid.uuid4()}.mp3 await tools.call_tool(tts, textreply, output_pathaudio_path) result[audio_url] f/static/{os.path.basename(audio_path)} tasks[task_id] {status: success, result: result} except Exception as e: tasks[task_id] {status: failed, error: str(e)} app.get(/api/v1/multimodal/output/{task_id}) async def get_output(task_id: str): task tasks.get(task_id) if not task: return {code: 404, message: 任务不存在} if task[status] processing: return {code: 202, message: 处理中} if task[status] failed: return {code: 500, message: task[error]} return {code: 200, **task[result]} if __name__ __main__: os.makedirs(./tmp, exist_okTrue) os.makedirs(./output, exist_okTrue) import uvicorn uvicorn.run(app, host0.0.0.0, port8000)YOUR_MODEL_ID、YOUR_TTS_MODEL、YOUR_EMBEDDING_MODEL这些占位符都要替换成在 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_end 模型广场里确认过的真实模型 ID。不要凭印象填否则会在运行时收到模型不存在或路由错误。4. 运行、Swagger 验证与排障4.1 启动服务先确认环境变量里带上 API Key再启动服务export OPENAI_API_KEYYOUR_API_KEY python main.py如果你不想用环境变量也可以在代码里直接写成api_keyYOUR_API_KEY但要注意别把 Key 提交进 Git 仓库。启动后访问http://localhost:8000/docs能看到 FastAPI 自动生成的 Swagger 页面。4.2 在 Swagger 里完整跑一次多模态输入在 Swagger 找到POST /api/v1/multimodal/input接口按下面这样填参数session_id填test-session-001user_id填test-user-001text填“电饭煲煮饭跳闸后米饭还是夹生”image上传一张电饭煲面板按钮的故障截图audio上传一段 10 秒语音说“就是这个按钮按了没反应”timestamp填当前毫秒时间戳比如1717000000000output_preference填text点击 Execute 后如果返回结果是{code: 200, message: 请求已接收, task_id: ...}说明输入接口已经接通TaoToken 的兼容通道也正常响应了。然后拿返回的task_id调GET /api/v1/multimodal/output/{task_id}轮询几次后会看到code: 200以及最终的文本回复。整个链路里图片描述、prompt 推理、文本 embedding 全部走的是 TaoToken 这把 Key。4.3 常见报错与排查我们项目里实际遇到过的几类问题可以对照排查现象大概率原因处理方式返回 401 UnauthorizedAPI Key 无效或未设置打开 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_end 重新创建 Key替换YOUR_API_KEY返回 404 Not Foundbase_url拼接了/v1或模型 ID 不存在确认 SDK 里填的是https://taotoken.net/api模型 ID 以模型广场为准Swagger 上传图片后长时间不返回图片尺寸过大或本地 CLIP 推理耗时先把图片压缩到 1024×1024 以内再上传语音识别结果答非所问Whisper 对方言识别率低在 process_audio 里加一段方言转普通话预处理或换成更高精度的语音模型FAISS 检索报维度不匹配投影矩阵维度与实际模型输出维度不一致打印text_embedding、image_embedding、audio_embedding的实际 shape再初始化W_t、W_i、W_s最常见的还是第二个问题开发同学习惯性地把 Base URL 写成https://taotoken.net/api/v1结果 OpenAI SDK 自动拼成/v1/chat/completions后变成/v1/v1/chat/completions。记住一个原则官网地址带不带路径都只在浏览器里用SDK 的 Base URL 统一填https://taotoken.net/api。5. 多模态 Harness 落地的几个注意点5.1 模态优先级与记忆加权不同业务场景的优先级完全不同。电商售后场景里语音描述最具体我们给语音 8、图片 7、文本 5但如果换到安防场景摄像头抓拍图像的优先级应该最高。权重w1、w2、w3也最好按业务调整不要照搬默认的 0.3/0.5/0.2。记忆检索时可以加时间衰减最近 24 小时的图片特征权重设为 1.53 天前的文本特征权重降为 0.5这样当用户说“上次那个问题”时优先命中最近的故障截图而不是更早的无关文本。5.2 工具熔断与输出适配OCR、TTS 这类第三方能力很容易超时。ToolOrchestration 里要保留降级逻辑OCR 挂了就退回 GPT-4o 自带的图像理解能力TTS 挂了就返回文本。输出适配也要看用户画像老年用户优先返回语音加大字号文本程序员用户直接给文本加代码片段。5.3 这套 Harness 能复用在哪这次接通的 TaoToken Key 不只服务于售后客服。只要代码里统一走 OpenAI SDK 的clientInputAdapter 的图像描述可以换成工单识别ToolOrchestration 的 TTS 可以换成业务通知播报MultimodalMemory 可以换成图片/语音混合的 FAQ 知识库。Harness 层的五个组件不变变的只是接入的业务工具。如果你正准备把一套多模态售后 Agent 接到自己的业务里下一步建议先把这份代码跑通然后打开 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_end 注册并创建 API Key替换掉YOUR_API_KEY和那几个模型占位符再回到 Swagger 上传一次“文本 故障截图 语音”。看到task_id返回的那一刻图片描述、语音转写、GPT-4o 推理和输出转换就都收拢到同一条兼容通道里了。