与 AGI 相关的搜索热度一直很高而“多模态 AGI”又是当前讨论最集中的方向之一。如果只看热搜会以为 4 个月后要交付的是一个能像人类一样思考的完整智能体但在工程团队里这个目标必须被翻译成需求文档、接口协议、评测集和上线清单。本文把“奥特曼最后一战4个月后交付AGI”当作项目代号拆解一条可落地的交付路径用 4 个月时间交付一个能同时处理文本、图片、音频具备工具调用、知识检索、记忆能力和基础规划能力的多模态 AGI 应用。文中会给出项目结构、核心代码、参数选型、评测方法和排错链路适合正在做智能体应用、多模态问答或企业 AI 中台的开发者参考。1. 先把“最后一战”拆成可验收的工程目标1.1 警惕热词与工程目标的距离“AGI”这个词在讨论中经常被当成一个终极目标但落到工程里它并不是一个可以直接开发的需求。团队如果带着“我们要做一个 AGI”的预期进入开发很容易在方案评审阶段就吵起来模型选哪个、上下文长度多少、是否需要多智能体、算力够不够、评测怎么算通过。更稳妥的做法是把“多模态 AGI”定义为一个系统能力组合。它不需要无限趋近通用智能只要能在限定业务范围内同时完成感知、理解、决策、执行和记忆并且能处理多种输入模态就可以作为一个可交付的 AGI 应用版本。真正要交付的内容包含五个部分多模态感知文本、图片、音频、视频分帧等输入解析。理解与生成基于大模型生成回答、摘要、结构化数据。工具与行动通过 API、数据库、计算器等外部能力完成真实操作。记忆跨会话保存用户偏好、业务上下文和长期知识。规划与协同将复杂任务拆成子任务必要时交给不同角色完成。1.2 给“多模态 AGI 应用”画一张能力地图开发前先画能力地图是为了避免把力气花在模型技术本身而不是花在业务交付上。下面这个表格可以作为需求评审的起点层级能力名称验收示例感知层文本、图片、语音输入处理用户上传一张表格图片系统输出结构化 JSON理解层多模态回答与摘要根据产品图片和说明书回答“这个按钮有什么用”行动层工具调用与外部 API 操作用户问“北京今天适合穿什么”系统调用天气接口后给出建议记忆层会话记忆与长期记忆用户第二次登录时仍然记得上次项目的筛选条件规划层多任务拆分与多智能体协同用户说“整理本周周报并发给李工”系统自动检索、生成并发送这五层并不是全部都会在第一个月做出来而是作为最终验收范围。每一层都要有独立评测指标例如图片解析准确率、工具调用成功率、记忆回正率、多步任务完成率。1.3 四个月路线图怎么安排四个月看起来很紧但如果把范围控制在一个业务域内时间仍然够用。关键在于阶段目标要清晰、每个阶段必须有可运行产物。阶段时间目标交付物第一月第 1 至 4 周跑通文本对话和工具调用最小闭环一个可对话、可调用天气/计算器接口的服务第二月第 5 至 8 周接入图片、语音接入记忆多模态输入可回答业务问题能记住用户偏好第三月第 9 至 12 周接入知识库检索、多智能体编排、评测和安全护栏具备 RAG 能力和评测报告线上拦截明显风险第四月第 13 至 16 周压测、监控、上线演练、复盘生产可用版本、部署文档、排错手册阶段划分的正确顺序是先打通窄链路再扩展宽度。不要一开始就上多智能体因为多智能体的效果依赖单智能体是否稳定也不要一开始就追求长上下文因为长上下文会放大记忆和成本问题。2. 环境准备与项目骨架先把最小服务跑起来2.1 运行环境和依赖版本要对齐不同环境中出现“本地能跑、服务器不能跑”的问题最常见的来源是依赖版本不一致。多模态项目涉及图片处理、音频转换、向量检索和大模型 SDK建议先把版本固定下来。下面是本文示例使用的推荐环境组件推荐版本说明Python3.10 以上类型语法和异步能力更友好FastAPI0.110 以上用于提供 HTTP 接口Uvicorn0.29 以上ASGI 服务器Pydantic2.x参数校验和配置管理openai1.x兼容 OpenAI 协议的大模型接口客户端python-multipart0.0.9 以上FastAPI 接收文件上传时需要Pillow10.x图片读取和压缩opencv-python4.x视频分帧和图像预处理按需安装faiss-cpu1.7.x向量检索学习环境用 CPU 版即可redis5.x会话记忆和缓存安装命令可以写进 requirements.txt不要直接裸装最新版pip install fastapi0.110,1.0 uvicorn[standard]0.29,1.0 \ pydantic2.0,3.0 openai1.0,2.0 \ python-multipart0.0.9 Pillow10.0,11.0 \ opencv-python4.8,5.0 faiss-cpu1.7,2.0 \ redis5.0,6.0如果原始项目已经使用了其他 LLM SDK落地前必须先确认模型接口是否兼容 OpenAI 协议。很多自建模型服务也提供兼容接口选择这类接口可以减少大量底层代码。2.2 项目目录结构多模态智能体项目建议按模块拆分不要把全部逻辑写在一个 main.py 里。下面是适合中期迭代的目录结构multimodal-agi/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 入口 │ ├── config.py # 环境变量与配置 │ ├── models.py # 请求/响应数据结构 │ ├── agent.py # 智能体主流程 │ ├── tools.py # 工具函数与工具注册 │ ├── memory.py # 记忆读写 │ ├── rag.py # 知识库检索 │ ├── guardrails.py # 输入和输出安全处理 │ └── utils/ │ ├── image_utils.py # 图片压缩与编码 │ └── audio_utils.py # 音频转写与分片 ├── cases/ # 评测样例 │ ├── t001_invoice.jpg │ └── evaluation.yaml ├── data/ │ └── knowledge/ # 知识库原始文件 ├── requirements.txt └── README.md这个结构的好处是每一层都有独立文件未来替换模型、增加工具、调整记忆策略时不需要重写整个项目。2.3 先启动一个最小多模态接口第一步不急着写完整智能体先让接口能接收文本和图片并返回一个固定结构的结果。这样能确认文件上传、图片解析、参数校验是否正确。from contextlib import asynccontextmanager from fastapi import FastAPI, UploadFile, File, Form from pydantic import BaseModel class ChatRequest(BaseModel): message: str user_id: str anonymous history: list[dict] [] image_bytes: bytes | None None asynccontextmanager async def lifespan(app: FastAPI): # 这里初始化模型客户端、向量库和日志句柄 yield app FastAPI(titlemultimodal-agi-demo, version0.1.0) app.post(/v1/chat) async def chat( message: str Form(...), user_id: str Form(anonymous), image: UploadFile | None File(defaultNone), ): image_bytes None if image and image.filename: image_bytes await image.read() request ChatRequest( messagemessage, user_iduser_id, image_bytesimage_bytes, ) # 先用固定结果验证链路后续替换成真正的 agent 逻辑 return { reply: freceive message{message}, image_size{len(image_bytes) if image_bytes else 0}, request_id: request.user_id, }保存为app/main.py后启动服务uvicorn app.main:app --host 0.0.0.0 --port 8000使用 curl 验证curl -X POST http://127.0.0.1:8000/v1/chat \ -F message你好 \ -F user_iddev-001 \ -F imagecases/t001_invoice.jpg正常返回{ reply: receive message你好, image_size24576, request_id: dev-001 }这一步的关键不是功能多丰富而是确认网络层和数据层没有低级错误。实际项目中图片上传后还要做大小限制、格式校验和敏感内容扫描不能直接丢给大模型。这里要特别注意不要在高频上传场景里把原始二进制直接写入数据库或 Redis建议先压缩再存储。3. 实现多模态输入、工具调用与记忆3.1 统一消息结构是智能体稳定的前提智能体本质上是在一个循环里反复处理“用户消息、模型返回、工具结果”。如果每个模块的消息结构都不一样后面接多智能体时就会非常痛苦。建议把消息统一成大模型接口常见的格式字段类型说明rolestringsystem、user、assistant、toolcontentstring or list文本内容或包含图片/音频的多模态内容tool_call_idstring工具调用时关联返回结果namestring工具名称核心数据结构可以这样定义from typing import Literal, Optional from pydantic import BaseModel, Field class ChatMessage(BaseModel): role: Literal[system, user, assistant, tool] content: str | list tool_call_id: Optional[str] None name: Optional[str] None class ChatRequest(BaseModel): message: str user_id: str anonymous history: list[ChatMessage] [] image_bytes: bytes | None None audio_bytes: bytes | None None request_id: Optional[str] Field(defaultNone, description链路追踪ID)统一结构后无论传入的是文本、图片还是工具结果都能用同一套消息列表驱动模型。3.2 图片和音频要先预处理不能直接照搬原始文件大部分模型对图片的输入有尺寸、格式和 token 限制。真实场景中用户上传的图片可能是 10 MB 的高清照片直接塞给模型不仅慢而且容易超过接口限制。推荐先做三步处理读取图片并转换为 RGB。限制最长边例如 2048 像素。转成 JPEG 并压缩质量再转 base64。import base64 from io import BytesIO from PIL import Image def image_bytes_to_content(image_bytes: bytes, max_side: int 2048) - dict: image Image.open(BytesIO(image_bytes)) if image.mode ! RGB: image image.convert(RGB) image.thumbnail((max_side, max_side)) buf BytesIO() image.save(buf, formatJPEG, quality85) base64_data base64.b64encode(buf.getvalue()).decode(utf-8) return { type: image, image_url: {url: fdata:image/jpeg;base64,{base64_data}}, }音频的处理思路类似先转成模型支持的格式和采样率再传送给语音转写接口。如果模型接口本身支持音频 base64也要控制时长超长音频先分片再汇总转写结果。这里很容易踩坑压缩后图片信息丢失导致 OCR 和细粒度识别失败。所以在实际业务里图片压缩参数要针对场景测试。例如发票识别要求清晰压缩质量就不要低于 85封面图识别则可以压缩到 70。3.3 工具调用循环要设置轮数上限多模态智能体如果没有工具调用就只能做“聊天”不能做“交付”。工具调用让模型可以查询数据库、调用天气接口、执行计算。实现工具调用的核心是一个循环把用户请求和工具定义发给模型。模型返回文本或返回一个工具调用请求。如果返回工具调用请求执行对应函数。把工具结果追加到消息列表。再次请求模型直到模型返回最终文本或达到最大轮数。下面是一个最小实现import json from openai import OpenAI client OpenAI() WEATHER_TOOL { type: function, function: { name: get_weather, description: 查询指定城市当前天气, parameters: { type: object, properties: { city: {type: string, description: 城市名例如 北京} }, required: [city], }, }, } def get_weather(city: str) - str: # 真实项目中替换为天气服务 API return json.dumps({city: city, weather: 晴, temperature: 24}, ensure_asciiFalse) def run_tool(name: str, arguments: str) - str: args json.loads(arguments) if name get_weather: return get_weather(**args) return json.dumps({error: funknown tool: {name}}) messages [ {role: system, content: 你是多模态助手可以调用工具完成用户请求。}, {role: user, content: 北京现在天气怎么样}, ] # 工具调用上限设置为 5防止模型反复调用导致死循环 for step in range(5): response client.chat.completions.create( modelyour-model-name, messagesmessages, tools[WEATHER_TOOL], ) msg response.choices[0].message if msg.tool_calls: messages.append(msg) for tool_call in msg.tool_calls: tool_result run_tool( tool_call.function.name, tool_call.function.arguments, ) messages.append( { role: tool, tool_call_id: tool_call.id, content: tool_result, } ) continue messages.append(msg) print(msg.content) break实际生产项目中工具函数要有独立注册表不要用大量 if/else 堆工具否则新增工具时改动面很大。工具执行还要加上超时、重试和异常捕获不能让单个工具拖垮整个智能体。3.4 记忆层要分短期和长期记忆是“交付 AGI”体验的重要一环。没有记忆的系统每次对话都像第一次见面有记忆的系统用户重复提问时会感受到明显差异。建议分两层短期记忆保存在 Redis 里TTL 设置为 1 到 24 小时存储当前会话上下文。长期记忆存储在向量库或关系型数据库里保存用户偏好、历史结论和常用参数。短期记忆实现示例import json import redis r redis.Redis(hostlocalhost, port6379, db0) def save_session(user_id: str, messages: list[dict], ttl: int 3600): key fsession:{user_id} r.set(key, json.dumps(messages, ensure_asciiFalse), exttl) def load_session(user_id: str) - list[dict]: key fsession:{user_id} raw r.get(key) if not raw: return [] return json.loads(raw)长期记忆要经过提取和确认不能一股脑把所有对话都写入。建议只保存用户明确表达过的偏好例如“我习惯用 Excel 导出”“价格低于 1000 元才推送”。这些内容格式化成键值对或句子写入向量库后在每次对话开始前召回。4. 接入知识库、多智能体和业务数据4.1 用 RAG 解决“模型没见过你的业务数据”问题大模型训练数据通常不包含企业内部知识。想让智能体回答公司制度、产品参数、私有文档内容就需要 RAG。RAG 的基本流程是切分文档、生成向量、检索相似片段、把片段拼进提示词。切分文档时要注意不要按固定字符数硬切推荐按段落或标题先分块再控制块大小。def chunk_text(text: str, chunk_size: int 500, overlap: int 50) - list[str]: if len(text) chunk_size: return [text] chunks [] start 0 while start len(text): end min(start chunk_size, len(text)) chunk text[start:end] if chunk: chunks.append(chunk) if end len(text): break start end - overlap return chunks生成向量后可以用 FAISS 建立索引import faiss import numpy as np class VectorStore: def __init__(self, dim: int 1024): self.index faiss.IndexFlatIP(dim) self.texts [] def add(self, text: str, embedding: list[float]): self.index.add(np.array([embedding], dtypefloat32)) self.texts.append(text) def search(self, embedding: list[float], top_k: int 5) - list[str]: scores, indices self.index.search( np.array([embedding], dtypefloat32), top_k, ) return [self.texts[i] for i in indices[0]]检索召回后把片段按固定模板拼进提示词并要求模型只依据片段回答。如果没有相关内容模型应该回答“知识库中未找到”而不是自己编造。这个约束非常重要否则 RAG 会变成“模型幻觉的高级触发器”。4.2 多智能体编排不是必须的但要做就做“路由器执行器”多模态 AGI 项目做到第三个月容易产生“要用多智能体”的冲动。多智能体能解决复杂任务但也会带来不稳定、成本高、链路难排查的问题。如果要用建议先采用“路由器执行器”模式一个主控智能体负责理解用户目标。根据目标选择执行智能体例如检索智能体、工具智能体、写作者。执行智能体返回结果后主控智能体汇总输出。可以用提示词做一个轻量路由器def select_agent(task: str) - str: if 查 in task or 文档 in task or 知识 in task: return retrieval if 计算 in task or 天气 in task: return tool return default生产环境不建议只靠一段 Python if 做路由可以换成结构化提示词让模型输出 JSON{ intent: retrieval, task: 查找公司年假制度, parameters: { keyword: 年假 } }多智能体有一个重要注意点每个智能体都要有独立的上下文窗口预算。不能让主控智能体把全部子任务结果一次性塞进去否则上下文很快超限。建议每个执行智能体只返回结构化摘要由主控汇总。4.3 业务数据对接要画边界智能体如果要操作业务系统比如查订单、改审批状态、发邮件必须设置边界。最核心的边界是只读数据建议由智能体直接查询。写操作必须经过用户二次确认。涉及删除、转账、对外发送内容时必须有独立审批。对外接口需要在网关层做权限校验不能依赖提示词约束。可以设计一个统一的工具元信息表工具名称类型是否需要确认超时风险等级query_order只读否5s低export_report只读否30s低send_email写操作是10s高delete_record写操作是5s高危“提示词里约定不要删除数据”不是安全方案。高危工具必须在代码层面加上确认参数只有用户显式输入“确认发送”时才执行。5. 评测、安全护栏与可观测性5.1 没有评测集效果优化就是凭感觉很多团队在开发智能体时只靠人工试几个问题就判断“效果不错”。这样的判断无法支撑 4 个月持续迭代。必须从第二个月开始建立离线评测集。评测样例可以直接用 YAML 管理cases: - id: t001 input: text: 这张图片里的发票总额是多少 image: cases/t001_invoice.jpg expect: field: total_amount equals: 120.00 not_contains: - error - 无法识别 - id: t002 input: text: 北京今天天气适合穿什么 expect: contains: - 温度 - 穿衣 tools_used: - get_weather评测时可以计算以下指标指标计算方式示例工具调用准确率正确调用工具数 / 总工具调用次数90%回答命中率回答中包含关键信息数 / 用例数85%幻觉率回答中出现知识库没有的内容 / 总回答数低于 5%完整执行率多步任务全部完成数 / 多步任务总数70%平均响应时间从请求到返回的总耗时3s离线评测报告要落到文件里每次模型或提示词改动后重新跑一遍避免“改好了 A 问题弄坏了 B 问题”的情况。5.2 输入和输出安全护栏必须放在代码层多模态输入带来了更多安全风险。图片里可以嵌入文字攻击音频里可以包含诱导指令。不要依赖“模型不会遵循恶意指令”这种假设。输入侧至少做三件事检查文件类型和大小禁止非白名单格式。对文本进行敏感词、Prompt 注入检测。图片先过 OCR 或视觉理解检测是否存在异常指令。输出侧至少做两件事过滤个人敏感信息例如身份证号、手机号。对高危工具的调用结果做脱敏后再展示。一个简单的敏感信息检查函数import re PHONE_PATTERN re.compile(r1[3-9]\d{9}) def mask_sensitive(text: str) - str: return PHONE_PATTERN.sub(lambda m: m.group(0)[:3] **** m.group(0)[-4:], text)这些规则虽然简单但能挡住绝大多数“无意泄露”。真正的恶意攻击还需要专门的红队测试因此 4 个月项目中至少要安排两周做安全测试。5.3 日志、追踪和监控要提前埋点多模态智能体链路很长文件解析、向量检索、模型推理、工具调用、记忆读写任何一环出了问题都可能导致最终结果异常。如果日志里只有一句replyxxx出了问题很难定位。建议从第一天开始就在日志中打印request_iduser_id输入消息摘要每个环节耗时模型名称和 token 数工具调用参数和结果异常堆栈日志输出推荐 JSON 格式方便采集到 ELK 或 Loki 中import json import logging import time logger logging.getLogger(multimodal-agi) def log_with_context(request_id: str, event: str, data: dict): logger.info( json.dumps( {request_id: request_id, event: event, ts: time.time(), **data}, ensure_asciiFalse, ) )如果是生产环境还需要接入分布式追踪在模型调用、向量检索、Redis 读写等关键节点记录 span。否则一次多步任务耗时 15 秒根本无法肉眼判断瓶颈在哪。6. 常见坑与排查链路6.1 多模态文件解析失败现象接口返回 500或者模型回复“我看不到图片”。可能原因图片格式不在白名单中。图片过大base64 后超出模型接口限制。音频格式不是模型支持的格式。上传时字段名没写对导致文件没进入解析逻辑。排查顺序先打印文件名称、大小和 Content-Type。用 Python 手动打开文件确认能否读取。检查 base64 编码是否完整。检查模型接口是否对分辨率、时长、token 数有限制。用最小图片测试逐步增加复杂度。解决方案上传入口统一做格式校验。图片统一压缩为 JPEG。音频统一转成模型支持的采样率和编码。6.2 工具调用出现死循环或者反复调用同一工具现象模型不停调用天气接口或调用同一个工具 5 次后仍然不返回最终答案。可能原因工具返回的结果没有真实解决用户问题。工具定义描述不清晰模型不知道何时停止。工具结果格式错误模型无法解析。循环没有设置最大轮数。排查方式在循环里打印每一步 messages 长度和 tool_call 参数。检查工具返回内容是否符合模型期望的 JSON。查看模型是否因为缺少字段而持续追问。解决方案给工具函数增加自动纠正能力例如城市名归一化。在系统提示词里说明“调用工具后如果获得结果立即组织最终回答”。严格限制工具调用最大轮数为 3 到 5。问题现象常见原因检查方式处理建议工具一直反复调用工具结果没有命中用户意图打印工具入参和结果增加工具描述归一化入参工具结果无法解析返回 JSON 不规范检查工具返回字符串用 json.dumps 强制序列化卡在循环结束没有最终回答分支处理查看循环 return 逻辑设置 max_steps超限返回兜底文案6.3 长上下文很快被撑爆现象对话十几轮后请求报错提示 tokens 超限或响应越来越慢。可能原因历史消息全部无差别塞进上下文。工具调用结果太长占用了大量 token。图片每次以 base64 原图进入上下文。解决方案历史消息做滑窗例如只保留最近 10 轮。工具结果先做摘要再放回消息列表。图片先压缩第二月再加视觉原文重述环节避免重复传递大图。给每个 agent 单独设置 token 预算。推荐做法MAX_HISTORY_ROUNDS 10 MAX_TOOL_RESULT_LENGTH 800 def trim_history(history: list[dict]) - list[dict]: if len(history) MAX_HISTORY_ROUNDS * 2: return history return history[-(MAX_HISTORY_ROUNDS * 2):] def trim_tool_result(text: str) - str: if len(text) MAX_TOOL_RESULT_LENGTH: return text return text[:MAX_TOOL_RESULT_LENGTH] ...(truncated)6.4 离线评测通过线上效果很差现象yaml 评测集通过率 90%但用户反馈经常答非所问。可能原因线上真实请求的表述和评测集差距大。线上图片质量不稳定评测图片太干净。知识库更新后没有重建索引。线上请求量高系统降级为短文本模型或低参数模型。排查方式从线上日志抽取一周真实用户问题扩充评测集。检查线上使用的模型名称、温度参数、上下文窗口是否与评测环境一致。检查知识库文档版本和向量索引时间。预防建议每周从真实日志补充 20 到 50 条评测样例。上线前做 A/B 对比。知识库更新后必须重建或增量更新向量索引。7. 落地检查清单与下一步扩展7.1 4 个月交付前的最小检查清单在正式宣布“交付多模态 AGI”之前建议逐项检查下面这个清单[ ] 文本对话、图片识别、语音转写三条链路是否全部有接口测试样例。[ ] 工具调用是否设置最大轮数、超时和异常兜底。[ ] 会话记忆是否能在服务重启后恢复。[ ] 长期记忆是否只保存用户明确表达的信息。[ ] 知识库是否有版本管理更新后是否触发向量索引重建。[ ] 离线评测集是否覆盖图片、音频、工具调用、多步任务四类场景。[ ] 输出侧是否做了敏感信息脱敏和高危操作确认。[ ] 日志中是否包含 request_id、环节耗时、工具参数、模型 token 数。[ ] 是否进行了压测确认并发场景下不会因为图片上传和向量检索拖垮服务。[ ] 是否编写了排错手册覆盖“文件解析失败”“上下文超限”“工具循环”“评测通过线上失败”四类问题。这个清单不是形式主义。每一个条目都对应一类真实发生过的问题。如果清单里有超过两项未完成建议先不要对外宣称“交付 AGI”而是继续补齐工程底座。7.2 从多模态应用向“更像 AGI”的方向扩展第一版交付完成后可以沿着四个方向扩展方向具体动作需要新增的基础设施更复杂的规划加入任务依赖图支持分支和并行任务工作流引擎、任务状态存储更统一的记忆用户画像、事件记忆、业务实体记忆三层建模图数据库或关系表、定时任务更强的多模态视频理解、文档版面恢复、图表问答视频分帧服务、OCR 服务、表格结构解析更可靠的行动浏览器自动化、审批流对接、消息发送权限中心、审计日志、人机确认页面每次扩展都要遵循同一个原则先补评测集再改代码。否则能力越加越多效果越来越难评估。7.3 如何判断“最后一战”是否真的完成“4 个月后交付 AGI”这句口号真正的价值不是定义“AGI”而是给团队定了一个时间盒。时间盒让人必须做取舍哪些能力先不做哪些环节必须自动化哪些风险可以通过评测和护栏兜住。一个更现实的完成标准是用户可以用自然语言完成原本需要多次点击的操作。系统能理解图片和语音输入并返回可信结果。系统能调用外部工具完成真实动作。高危操作有确认和审计。所有关键路径可以被日志和评测数据追踪。做到这些哪怕没有达到“完整通用人工智能”的定义也已经形成了一个真正可以落地、可以迭代、可以对外展示的多模态 AGI 应用。下一阶段的能力演进都建立在这次交付的工程底座之上。
多模态AGI应用交付实战：4个月从零到工程落地
与 AGI 相关的搜索热度一直很高而“多模态 AGI”又是当前讨论最集中的方向之一。如果只看热搜会以为 4 个月后要交付的是一个能像人类一样思考的完整智能体但在工程团队里这个目标必须被翻译成需求文档、接口协议、评测集和上线清单。本文把“奥特曼最后一战4个月后交付AGI”当作项目代号拆解一条可落地的交付路径用 4 个月时间交付一个能同时处理文本、图片、音频具备工具调用、知识检索、记忆能力和基础规划能力的多模态 AGI 应用。文中会给出项目结构、核心代码、参数选型、评测方法和排错链路适合正在做智能体应用、多模态问答或企业 AI 中台的开发者参考。1. 先把“最后一战”拆成可验收的工程目标1.1 警惕热词与工程目标的距离“AGI”这个词在讨论中经常被当成一个终极目标但落到工程里它并不是一个可以直接开发的需求。团队如果带着“我们要做一个 AGI”的预期进入开发很容易在方案评审阶段就吵起来模型选哪个、上下文长度多少、是否需要多智能体、算力够不够、评测怎么算通过。更稳妥的做法是把“多模态 AGI”定义为一个系统能力组合。它不需要无限趋近通用智能只要能在限定业务范围内同时完成感知、理解、决策、执行和记忆并且能处理多种输入模态就可以作为一个可交付的 AGI 应用版本。真正要交付的内容包含五个部分多模态感知文本、图片、音频、视频分帧等输入解析。理解与生成基于大模型生成回答、摘要、结构化数据。工具与行动通过 API、数据库、计算器等外部能力完成真实操作。记忆跨会话保存用户偏好、业务上下文和长期知识。规划与协同将复杂任务拆成子任务必要时交给不同角色完成。1.2 给“多模态 AGI 应用”画一张能力地图开发前先画能力地图是为了避免把力气花在模型技术本身而不是花在业务交付上。下面这个表格可以作为需求评审的起点层级能力名称验收示例感知层文本、图片、语音输入处理用户上传一张表格图片系统输出结构化 JSON理解层多模态回答与摘要根据产品图片和说明书回答“这个按钮有什么用”行动层工具调用与外部 API 操作用户问“北京今天适合穿什么”系统调用天气接口后给出建议记忆层会话记忆与长期记忆用户第二次登录时仍然记得上次项目的筛选条件规划层多任务拆分与多智能体协同用户说“整理本周周报并发给李工”系统自动检索、生成并发送这五层并不是全部都会在第一个月做出来而是作为最终验收范围。每一层都要有独立评测指标例如图片解析准确率、工具调用成功率、记忆回正率、多步任务完成率。1.3 四个月路线图怎么安排四个月看起来很紧但如果把范围控制在一个业务域内时间仍然够用。关键在于阶段目标要清晰、每个阶段必须有可运行产物。阶段时间目标交付物第一月第 1 至 4 周跑通文本对话和工具调用最小闭环一个可对话、可调用天气/计算器接口的服务第二月第 5 至 8 周接入图片、语音接入记忆多模态输入可回答业务问题能记住用户偏好第三月第 9 至 12 周接入知识库检索、多智能体编排、评测和安全护栏具备 RAG 能力和评测报告线上拦截明显风险第四月第 13 至 16 周压测、监控、上线演练、复盘生产可用版本、部署文档、排错手册阶段划分的正确顺序是先打通窄链路再扩展宽度。不要一开始就上多智能体因为多智能体的效果依赖单智能体是否稳定也不要一开始就追求长上下文因为长上下文会放大记忆和成本问题。2. 环境准备与项目骨架先把最小服务跑起来2.1 运行环境和依赖版本要对齐不同环境中出现“本地能跑、服务器不能跑”的问题最常见的来源是依赖版本不一致。多模态项目涉及图片处理、音频转换、向量检索和大模型 SDK建议先把版本固定下来。下面是本文示例使用的推荐环境组件推荐版本说明Python3.10 以上类型语法和异步能力更友好FastAPI0.110 以上用于提供 HTTP 接口Uvicorn0.29 以上ASGI 服务器Pydantic2.x参数校验和配置管理openai1.x兼容 OpenAI 协议的大模型接口客户端python-multipart0.0.9 以上FastAPI 接收文件上传时需要Pillow10.x图片读取和压缩opencv-python4.x视频分帧和图像预处理按需安装faiss-cpu1.7.x向量检索学习环境用 CPU 版即可redis5.x会话记忆和缓存安装命令可以写进 requirements.txt不要直接裸装最新版pip install fastapi0.110,1.0 uvicorn[standard]0.29,1.0 \ pydantic2.0,3.0 openai1.0,2.0 \ python-multipart0.0.9 Pillow10.0,11.0 \ opencv-python4.8,5.0 faiss-cpu1.7,2.0 \ redis5.0,6.0如果原始项目已经使用了其他 LLM SDK落地前必须先确认模型接口是否兼容 OpenAI 协议。很多自建模型服务也提供兼容接口选择这类接口可以减少大量底层代码。2.2 项目目录结构多模态智能体项目建议按模块拆分不要把全部逻辑写在一个 main.py 里。下面是适合中期迭代的目录结构multimodal-agi/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 入口 │ ├── config.py # 环境变量与配置 │ ├── models.py # 请求/响应数据结构 │ ├── agent.py # 智能体主流程 │ ├── tools.py # 工具函数与工具注册 │ ├── memory.py # 记忆读写 │ ├── rag.py # 知识库检索 │ ├── guardrails.py # 输入和输出安全处理 │ └── utils/ │ ├── image_utils.py # 图片压缩与编码 │ └── audio_utils.py # 音频转写与分片 ├── cases/ # 评测样例 │ ├── t001_invoice.jpg │ └── evaluation.yaml ├── data/ │ └── knowledge/ # 知识库原始文件 ├── requirements.txt └── README.md这个结构的好处是每一层都有独立文件未来替换模型、增加工具、调整记忆策略时不需要重写整个项目。2.3 先启动一个最小多模态接口第一步不急着写完整智能体先让接口能接收文本和图片并返回一个固定结构的结果。这样能确认文件上传、图片解析、参数校验是否正确。from contextlib import asynccontextmanager from fastapi import FastAPI, UploadFile, File, Form from pydantic import BaseModel class ChatRequest(BaseModel): message: str user_id: str anonymous history: list[dict] [] image_bytes: bytes | None None asynccontextmanager async def lifespan(app: FastAPI): # 这里初始化模型客户端、向量库和日志句柄 yield app FastAPI(titlemultimodal-agi-demo, version0.1.0) app.post(/v1/chat) async def chat( message: str Form(...), user_id: str Form(anonymous), image: UploadFile | None File(defaultNone), ): image_bytes None if image and image.filename: image_bytes await image.read() request ChatRequest( messagemessage, user_iduser_id, image_bytesimage_bytes, ) # 先用固定结果验证链路后续替换成真正的 agent 逻辑 return { reply: freceive message{message}, image_size{len(image_bytes) if image_bytes else 0}, request_id: request.user_id, }保存为app/main.py后启动服务uvicorn app.main:app --host 0.0.0.0 --port 8000使用 curl 验证curl -X POST http://127.0.0.1:8000/v1/chat \ -F message你好 \ -F user_iddev-001 \ -F imagecases/t001_invoice.jpg正常返回{ reply: receive message你好, image_size24576, request_id: dev-001 }这一步的关键不是功能多丰富而是确认网络层和数据层没有低级错误。实际项目中图片上传后还要做大小限制、格式校验和敏感内容扫描不能直接丢给大模型。这里要特别注意不要在高频上传场景里把原始二进制直接写入数据库或 Redis建议先压缩再存储。3. 实现多模态输入、工具调用与记忆3.1 统一消息结构是智能体稳定的前提智能体本质上是在一个循环里反复处理“用户消息、模型返回、工具结果”。如果每个模块的消息结构都不一样后面接多智能体时就会非常痛苦。建议把消息统一成大模型接口常见的格式字段类型说明rolestringsystem、user、assistant、toolcontentstring or list文本内容或包含图片/音频的多模态内容tool_call_idstring工具调用时关联返回结果namestring工具名称核心数据结构可以这样定义from typing import Literal, Optional from pydantic import BaseModel, Field class ChatMessage(BaseModel): role: Literal[system, user, assistant, tool] content: str | list tool_call_id: Optional[str] None name: Optional[str] None class ChatRequest(BaseModel): message: str user_id: str anonymous history: list[ChatMessage] [] image_bytes: bytes | None None audio_bytes: bytes | None None request_id: Optional[str] Field(defaultNone, description链路追踪ID)统一结构后无论传入的是文本、图片还是工具结果都能用同一套消息列表驱动模型。3.2 图片和音频要先预处理不能直接照搬原始文件大部分模型对图片的输入有尺寸、格式和 token 限制。真实场景中用户上传的图片可能是 10 MB 的高清照片直接塞给模型不仅慢而且容易超过接口限制。推荐先做三步处理读取图片并转换为 RGB。限制最长边例如 2048 像素。转成 JPEG 并压缩质量再转 base64。import base64 from io import BytesIO from PIL import Image def image_bytes_to_content(image_bytes: bytes, max_side: int 2048) - dict: image Image.open(BytesIO(image_bytes)) if image.mode ! RGB: image image.convert(RGB) image.thumbnail((max_side, max_side)) buf BytesIO() image.save(buf, formatJPEG, quality85) base64_data base64.b64encode(buf.getvalue()).decode(utf-8) return { type: image, image_url: {url: fdata:image/jpeg;base64,{base64_data}}, }音频的处理思路类似先转成模型支持的格式和采样率再传送给语音转写接口。如果模型接口本身支持音频 base64也要控制时长超长音频先分片再汇总转写结果。这里很容易踩坑压缩后图片信息丢失导致 OCR 和细粒度识别失败。所以在实际业务里图片压缩参数要针对场景测试。例如发票识别要求清晰压缩质量就不要低于 85封面图识别则可以压缩到 70。3.3 工具调用循环要设置轮数上限多模态智能体如果没有工具调用就只能做“聊天”不能做“交付”。工具调用让模型可以查询数据库、调用天气接口、执行计算。实现工具调用的核心是一个循环把用户请求和工具定义发给模型。模型返回文本或返回一个工具调用请求。如果返回工具调用请求执行对应函数。把工具结果追加到消息列表。再次请求模型直到模型返回最终文本或达到最大轮数。下面是一个最小实现import json from openai import OpenAI client OpenAI() WEATHER_TOOL { type: function, function: { name: get_weather, description: 查询指定城市当前天气, parameters: { type: object, properties: { city: {type: string, description: 城市名例如 北京} }, required: [city], }, }, } def get_weather(city: str) - str: # 真实项目中替换为天气服务 API return json.dumps({city: city, weather: 晴, temperature: 24}, ensure_asciiFalse) def run_tool(name: str, arguments: str) - str: args json.loads(arguments) if name get_weather: return get_weather(**args) return json.dumps({error: funknown tool: {name}}) messages [ {role: system, content: 你是多模态助手可以调用工具完成用户请求。}, {role: user, content: 北京现在天气怎么样}, ] # 工具调用上限设置为 5防止模型反复调用导致死循环 for step in range(5): response client.chat.completions.create( modelyour-model-name, messagesmessages, tools[WEATHER_TOOL], ) msg response.choices[0].message if msg.tool_calls: messages.append(msg) for tool_call in msg.tool_calls: tool_result run_tool( tool_call.function.name, tool_call.function.arguments, ) messages.append( { role: tool, tool_call_id: tool_call.id, content: tool_result, } ) continue messages.append(msg) print(msg.content) break实际生产项目中工具函数要有独立注册表不要用大量 if/else 堆工具否则新增工具时改动面很大。工具执行还要加上超时、重试和异常捕获不能让单个工具拖垮整个智能体。3.4 记忆层要分短期和长期记忆是“交付 AGI”体验的重要一环。没有记忆的系统每次对话都像第一次见面有记忆的系统用户重复提问时会感受到明显差异。建议分两层短期记忆保存在 Redis 里TTL 设置为 1 到 24 小时存储当前会话上下文。长期记忆存储在向量库或关系型数据库里保存用户偏好、历史结论和常用参数。短期记忆实现示例import json import redis r redis.Redis(hostlocalhost, port6379, db0) def save_session(user_id: str, messages: list[dict], ttl: int 3600): key fsession:{user_id} r.set(key, json.dumps(messages, ensure_asciiFalse), exttl) def load_session(user_id: str) - list[dict]: key fsession:{user_id} raw r.get(key) if not raw: return [] return json.loads(raw)长期记忆要经过提取和确认不能一股脑把所有对话都写入。建议只保存用户明确表达过的偏好例如“我习惯用 Excel 导出”“价格低于 1000 元才推送”。这些内容格式化成键值对或句子写入向量库后在每次对话开始前召回。4. 接入知识库、多智能体和业务数据4.1 用 RAG 解决“模型没见过你的业务数据”问题大模型训练数据通常不包含企业内部知识。想让智能体回答公司制度、产品参数、私有文档内容就需要 RAG。RAG 的基本流程是切分文档、生成向量、检索相似片段、把片段拼进提示词。切分文档时要注意不要按固定字符数硬切推荐按段落或标题先分块再控制块大小。def chunk_text(text: str, chunk_size: int 500, overlap: int 50) - list[str]: if len(text) chunk_size: return [text] chunks [] start 0 while start len(text): end min(start chunk_size, len(text)) chunk text[start:end] if chunk: chunks.append(chunk) if end len(text): break start end - overlap return chunks生成向量后可以用 FAISS 建立索引import faiss import numpy as np class VectorStore: def __init__(self, dim: int 1024): self.index faiss.IndexFlatIP(dim) self.texts [] def add(self, text: str, embedding: list[float]): self.index.add(np.array([embedding], dtypefloat32)) self.texts.append(text) def search(self, embedding: list[float], top_k: int 5) - list[str]: scores, indices self.index.search( np.array([embedding], dtypefloat32), top_k, ) return [self.texts[i] for i in indices[0]]检索召回后把片段按固定模板拼进提示词并要求模型只依据片段回答。如果没有相关内容模型应该回答“知识库中未找到”而不是自己编造。这个约束非常重要否则 RAG 会变成“模型幻觉的高级触发器”。4.2 多智能体编排不是必须的但要做就做“路由器执行器”多模态 AGI 项目做到第三个月容易产生“要用多智能体”的冲动。多智能体能解决复杂任务但也会带来不稳定、成本高、链路难排查的问题。如果要用建议先采用“路由器执行器”模式一个主控智能体负责理解用户目标。根据目标选择执行智能体例如检索智能体、工具智能体、写作者。执行智能体返回结果后主控智能体汇总输出。可以用提示词做一个轻量路由器def select_agent(task: str) - str: if 查 in task or 文档 in task or 知识 in task: return retrieval if 计算 in task or 天气 in task: return tool return default生产环境不建议只靠一段 Python if 做路由可以换成结构化提示词让模型输出 JSON{ intent: retrieval, task: 查找公司年假制度, parameters: { keyword: 年假 } }多智能体有一个重要注意点每个智能体都要有独立的上下文窗口预算。不能让主控智能体把全部子任务结果一次性塞进去否则上下文很快超限。建议每个执行智能体只返回结构化摘要由主控汇总。4.3 业务数据对接要画边界智能体如果要操作业务系统比如查订单、改审批状态、发邮件必须设置边界。最核心的边界是只读数据建议由智能体直接查询。写操作必须经过用户二次确认。涉及删除、转账、对外发送内容时必须有独立审批。对外接口需要在网关层做权限校验不能依赖提示词约束。可以设计一个统一的工具元信息表工具名称类型是否需要确认超时风险等级query_order只读否5s低export_report只读否30s低send_email写操作是10s高delete_record写操作是5s高危“提示词里约定不要删除数据”不是安全方案。高危工具必须在代码层面加上确认参数只有用户显式输入“确认发送”时才执行。5. 评测、安全护栏与可观测性5.1 没有评测集效果优化就是凭感觉很多团队在开发智能体时只靠人工试几个问题就判断“效果不错”。这样的判断无法支撑 4 个月持续迭代。必须从第二个月开始建立离线评测集。评测样例可以直接用 YAML 管理cases: - id: t001 input: text: 这张图片里的发票总额是多少 image: cases/t001_invoice.jpg expect: field: total_amount equals: 120.00 not_contains: - error - 无法识别 - id: t002 input: text: 北京今天天气适合穿什么 expect: contains: - 温度 - 穿衣 tools_used: - get_weather评测时可以计算以下指标指标计算方式示例工具调用准确率正确调用工具数 / 总工具调用次数90%回答命中率回答中包含关键信息数 / 用例数85%幻觉率回答中出现知识库没有的内容 / 总回答数低于 5%完整执行率多步任务全部完成数 / 多步任务总数70%平均响应时间从请求到返回的总耗时3s离线评测报告要落到文件里每次模型或提示词改动后重新跑一遍避免“改好了 A 问题弄坏了 B 问题”的情况。5.2 输入和输出安全护栏必须放在代码层多模态输入带来了更多安全风险。图片里可以嵌入文字攻击音频里可以包含诱导指令。不要依赖“模型不会遵循恶意指令”这种假设。输入侧至少做三件事检查文件类型和大小禁止非白名单格式。对文本进行敏感词、Prompt 注入检测。图片先过 OCR 或视觉理解检测是否存在异常指令。输出侧至少做两件事过滤个人敏感信息例如身份证号、手机号。对高危工具的调用结果做脱敏后再展示。一个简单的敏感信息检查函数import re PHONE_PATTERN re.compile(r1[3-9]\d{9}) def mask_sensitive(text: str) - str: return PHONE_PATTERN.sub(lambda m: m.group(0)[:3] **** m.group(0)[-4:], text)这些规则虽然简单但能挡住绝大多数“无意泄露”。真正的恶意攻击还需要专门的红队测试因此 4 个月项目中至少要安排两周做安全测试。5.3 日志、追踪和监控要提前埋点多模态智能体链路很长文件解析、向量检索、模型推理、工具调用、记忆读写任何一环出了问题都可能导致最终结果异常。如果日志里只有一句replyxxx出了问题很难定位。建议从第一天开始就在日志中打印request_iduser_id输入消息摘要每个环节耗时模型名称和 token 数工具调用参数和结果异常堆栈日志输出推荐 JSON 格式方便采集到 ELK 或 Loki 中import json import logging import time logger logging.getLogger(multimodal-agi) def log_with_context(request_id: str, event: str, data: dict): logger.info( json.dumps( {request_id: request_id, event: event, ts: time.time(), **data}, ensure_asciiFalse, ) )如果是生产环境还需要接入分布式追踪在模型调用、向量检索、Redis 读写等关键节点记录 span。否则一次多步任务耗时 15 秒根本无法肉眼判断瓶颈在哪。6. 常见坑与排查链路6.1 多模态文件解析失败现象接口返回 500或者模型回复“我看不到图片”。可能原因图片格式不在白名单中。图片过大base64 后超出模型接口限制。音频格式不是模型支持的格式。上传时字段名没写对导致文件没进入解析逻辑。排查顺序先打印文件名称、大小和 Content-Type。用 Python 手动打开文件确认能否读取。检查 base64 编码是否完整。检查模型接口是否对分辨率、时长、token 数有限制。用最小图片测试逐步增加复杂度。解决方案上传入口统一做格式校验。图片统一压缩为 JPEG。音频统一转成模型支持的采样率和编码。6.2 工具调用出现死循环或者反复调用同一工具现象模型不停调用天气接口或调用同一个工具 5 次后仍然不返回最终答案。可能原因工具返回的结果没有真实解决用户问题。工具定义描述不清晰模型不知道何时停止。工具结果格式错误模型无法解析。循环没有设置最大轮数。排查方式在循环里打印每一步 messages 长度和 tool_call 参数。检查工具返回内容是否符合模型期望的 JSON。查看模型是否因为缺少字段而持续追问。解决方案给工具函数增加自动纠正能力例如城市名归一化。在系统提示词里说明“调用工具后如果获得结果立即组织最终回答”。严格限制工具调用最大轮数为 3 到 5。问题现象常见原因检查方式处理建议工具一直反复调用工具结果没有命中用户意图打印工具入参和结果增加工具描述归一化入参工具结果无法解析返回 JSON 不规范检查工具返回字符串用 json.dumps 强制序列化卡在循环结束没有最终回答分支处理查看循环 return 逻辑设置 max_steps超限返回兜底文案6.3 长上下文很快被撑爆现象对话十几轮后请求报错提示 tokens 超限或响应越来越慢。可能原因历史消息全部无差别塞进上下文。工具调用结果太长占用了大量 token。图片每次以 base64 原图进入上下文。解决方案历史消息做滑窗例如只保留最近 10 轮。工具结果先做摘要再放回消息列表。图片先压缩第二月再加视觉原文重述环节避免重复传递大图。给每个 agent 单独设置 token 预算。推荐做法MAX_HISTORY_ROUNDS 10 MAX_TOOL_RESULT_LENGTH 800 def trim_history(history: list[dict]) - list[dict]: if len(history) MAX_HISTORY_ROUNDS * 2: return history return history[-(MAX_HISTORY_ROUNDS * 2):] def trim_tool_result(text: str) - str: if len(text) MAX_TOOL_RESULT_LENGTH: return text return text[:MAX_TOOL_RESULT_LENGTH] ...(truncated)6.4 离线评测通过线上效果很差现象yaml 评测集通过率 90%但用户反馈经常答非所问。可能原因线上真实请求的表述和评测集差距大。线上图片质量不稳定评测图片太干净。知识库更新后没有重建索引。线上请求量高系统降级为短文本模型或低参数模型。排查方式从线上日志抽取一周真实用户问题扩充评测集。检查线上使用的模型名称、温度参数、上下文窗口是否与评测环境一致。检查知识库文档版本和向量索引时间。预防建议每周从真实日志补充 20 到 50 条评测样例。上线前做 A/B 对比。知识库更新后必须重建或增量更新向量索引。7. 落地检查清单与下一步扩展7.1 4 个月交付前的最小检查清单在正式宣布“交付多模态 AGI”之前建议逐项检查下面这个清单[ ] 文本对话、图片识别、语音转写三条链路是否全部有接口测试样例。[ ] 工具调用是否设置最大轮数、超时和异常兜底。[ ] 会话记忆是否能在服务重启后恢复。[ ] 长期记忆是否只保存用户明确表达的信息。[ ] 知识库是否有版本管理更新后是否触发向量索引重建。[ ] 离线评测集是否覆盖图片、音频、工具调用、多步任务四类场景。[ ] 输出侧是否做了敏感信息脱敏和高危操作确认。[ ] 日志中是否包含 request_id、环节耗时、工具参数、模型 token 数。[ ] 是否进行了压测确认并发场景下不会因为图片上传和向量检索拖垮服务。[ ] 是否编写了排错手册覆盖“文件解析失败”“上下文超限”“工具循环”“评测通过线上失败”四类问题。这个清单不是形式主义。每一个条目都对应一类真实发生过的问题。如果清单里有超过两项未完成建议先不要对外宣称“交付 AGI”而是继续补齐工程底座。7.2 从多模态应用向“更像 AGI”的方向扩展第一版交付完成后可以沿着四个方向扩展方向具体动作需要新增的基础设施更复杂的规划加入任务依赖图支持分支和并行任务工作流引擎、任务状态存储更统一的记忆用户画像、事件记忆、业务实体记忆三层建模图数据库或关系表、定时任务更强的多模态视频理解、文档版面恢复、图表问答视频分帧服务、OCR 服务、表格结构解析更可靠的行动浏览器自动化、审批流对接、消息发送权限中心、审计日志、人机确认页面每次扩展都要遵循同一个原则先补评测集再改代码。否则能力越加越多效果越来越难评估。7.3 如何判断“最后一战”是否真的完成“4 个月后交付 AGI”这句口号真正的价值不是定义“AGI”而是给团队定了一个时间盒。时间盒让人必须做取舍哪些能力先不做哪些环节必须自动化哪些风险可以通过评测和护栏兜住。一个更现实的完成标准是用户可以用自然语言完成原本需要多次点击的操作。系统能理解图片和语音输入并返回可信结果。系统能调用外部工具完成真实动作。高危操作有确认和审计。所有关键路径可以被日志和评测数据追踪。做到这些哪怕没有达到“完整通用人工智能”的定义也已经形成了一个真正可以落地、可以迭代、可以对外展示的多模态 AGI 应用。下一阶段的能力演进都建立在这次交付的工程底座之上。