AI Agent 系统设计与多模态交互实验：从最小可用方案搭起

在设计 AI Agent 系统时，很多团队刚立项就试图构建一个复杂的 Multi-Agent 协作网络。他们把系统拆分成规划者（Planner）、执行者（Worker）、批评者（Critic）等五六个不同的智能体角色，甚至设计了繁琐的图状态路由。

然而在工程落地的初期，盲目增加 Agent 数量往往带来难以控制的非确定性与极高的 Token 耗费。真正的工程之道，是从一个单圈 ReAct 循环的最小可用方案（MVP）搭起，先把确定性防线打扎实。

1. 拒绝盲目复杂化：最小可用 AI Agent (MVP) 的关键组成

一个立竿见影的最小可用 Agent 系统，本质上只需要包含三个关键构件：

1. 确定的 Tool 注册表：暴露给 Agent 的工具数量要少而精，每个工具的参数 Schema 必须经由代码进行强类型校验。
2. 轻量级状态机（ReAct Loop）：一个控制“思考 -> 调工具 -> 观察结果 -> 总结”的单圈循环，并带有最大轮次硬限制。
3. 多模态输入裁剪器：对进入 Agent 的图像或长文档进行分辨率压缩与 Token 截断，防止单次请求直接将上下文压垮。

只要这三要素协同运转顺畅，Agent 就能解决 80% 的生产场景需求。

2. 单圈 ReAct 循环与确定性工具集配置

在 MVP 方案中，ReAct（Reasoning + Acting）循环必须是简单且透明的。

许多开源框架（如 LangChain 或 AutoGen）在底层封装了过多的抽象层，导致当 Agent 卡在某一步时，开发者根本查不到底层究竟向 API 发送了怎样的 Prompt 结构。

自己用几十行代码实现一个清爽的 ReAct 循环，不仅调试方便，还能精准把控每一次 Tool Call 的日志输出。

对于 Tool 的定义，要像定义 RESTful API 一样严谨：说明必须清晰直白，字段类型必须明确，并且每个 Tool 内部都要处理异常，不应把原生 Python Traceback 直接抛给 LLM。

3. 多模态交互处理：图像压缩、Token 消耗控制与上下文截断

在引入 Vision 多模态能力时，最大的隐形杀手是图像带来的 Token 爆炸。

一张原生的 4K 高清截图如果直接喂给大模型，后台可能需要将其切分为数十个 Patch，单张图片就消耗掉几千个 Token。

在 MVP 架构中，必须在 Payload 进入 ReAct 循环之前增加预处理模块：将图片长边等比例缩放至 1024px 以下，转换为 JPEG 格式并控制压缩质量（如 Quality=80）。这样做不仅能降低 70% 以上的 Token 消耗，还能大幅缩短模型的首包响应延迟（TTFT）。

4. MVP Agent 的轻量级 State Machine 代码实现

下面是一个不依赖任何繁重第三方 Agent 框架、用纯 Python 实现的最小可用多模态 Agent 方案。

包含完整的 Tool 注册机制、图像压缩预处理、ReAct 循环控制器以及死循环中断降级：

import io
import json
import asyncio
from typing import List, Dict, Any, Callable
from PIL import Image
import base64

# --- 1. 图像多模态预处理模块 ---
def compress_and_encode_image(image_bytes: bytes, max_dim: int = 1024) -> str:
    """
    对输入图像进行等比例缩放并压缩为 JPEG Base64 编码，控制 Token 预算
    """
    with Image.open(io.BytesIO(image_bytes)) as img:
        img = img.convert("RGB")
        width, height = img.size
        if max(width, height) > max_dim:
            scale = max_dim / float(max(width, height))
            new_size = (int(width * scale), int(height * scale))
            img = img.resize(new_size, Image.Resampling.LANCZOS)
        
        buffer = io.BytesIO()
        img.save(buffer, format="JPEG", quality=80)
        encoded_str = base64.b64encode(buffer.getvalue()).decode("utf-8")
        return f"data:image/jpeg;base64,{encoded_str}"

# --- 2. 确定性 Tool 注册表 ---
class ToolRegistry:
    def __init__(self):
        self._tools: Dict[str, Callable] = {}
        self._schemas: List[Dict[str, Any]] = []

    def register(self, name: str, description: str, param_schema: Dict[str, Any]):
        def decorator(func: Callable):
            self._tools[name] = func
            self._schemas.append({
                "name": name,
                "description": description,
                "parameters": param_schema
            })
            return func
        return decorator

    def execute(self, name: str, kwargs_json_str: str) -> str:
        if name not in self._tools:
            return f"Error: 找不到名为 '{name}' 的工具"
        try:
            kwargs = json.loads(kwargs_json_str) if isinstance(kwargs_json_str, str) else kwargs_json_str
            result = self._tools[name](**kwargs)
            return json.dumps(result, ensure_ascii=False)
        except Exception as ex:
            return f"Error: 执行工具 '{name}' 时抛出异常: {str(ex)}"

# 初始化工具库
tools = ToolRegistry()

@tools.register(
    name="query_inventory",
    description="查询仓库中某商品的库存数量与货位",
    param_schema={
        "type": "object",
        "properties": {"sku_id": {"type": "string", "description": "商品 SKU 编号"}},
        "required": ["sku_id"]
    }
)
def query_inventory(sku_id: str) -> Dict[str, Any]:
    # 真实场景接入数据库查询
    if sku_id == "SKU-998":
        return {"sku_id": sku_id, "stock": 42, "location": "A-03-12"}
    return {"sku_id": sku_id, "stock": 0, "location": "NONE"}

# --- 3. 轻量级 MVP ReAct Agent 引擎 ---
class MinimalReActAgent:
    def __init__(self, tool_registry: ToolRegistry, max_turns: int = 4):
        self.tools = tool_registry
        self.max_turns = max_turns

    async def run(self, user_text: str, image_bytes: Optional[bytes], llm_client_fn) -> str:
        messages = [
            {"role": "system", "content": "你是一个严谨的助手。根据需要调用工具解决问题，当收集到足够信息时直接给出最终回答。"}
        ]
        
        # 预处理多模态 Payload
        content_payload = [{"type": "text", "text": user_text}]
        if image_bytes:
            compressed_b64 = compress_and_encode_image(image_bytes)
            content_payload.append({"type": "image_url", "image_url": {"url": compressed_b64}})

        messages.append({"role": "user", "content": content_payload})

        for turn in range(self.max_turns):
            print(f"--- 启动 ReAct 轮次 {turn + 1}/{self.max_turns} ---")
            
            # 调用底层 LLM 模拟接口
            response = await llm_client_fn(messages, self.tools._schemas)
            
            # 检查 LLM 是否发起 Tool Calling 指令
            tool_call = response.get("tool_call")
            if not tool_call:
                # 终态：LLM 给出最终文本总结
                return response.get("content", "处理完成")

            tool_name = tool_call.get("name")
            tool_args = tool_call.get("arguments")
            print(f"Agent 发起 Tool 调起: {tool_name}, 参数: {tool_args}")

            # 在确定性环境下安全执行工具
            obs = self.tools.execute(tool_name, tool_args)
            print(f"Tool 返回 Observation: {obs}")

            # 将历史追加至上下文 Memory
            messages.append({"role": "assistant", "content": f"调用工具 {tool_name}"})
            messages.append({"role": "user", "content": f"工具 {tool_name} 输出观察结果: {obs}"})

        return "错误: Agent 达到最大允许轮次，已强制中止流程"

# 使用演练
if __name__ == "__main__":
    # 模拟多模态 LLM 客户端 API 行为
    async def mock_multimodal_llm(messages, tool_schemas):
        # 简单根据上一轮 Memory 模拟 ReAct 决策
        last_msg = str(messages[-1]["content"])
        if "SKU-998" in last_msg and "工具" not in last_msg:
            return {
                "tool_call": {
                    "name": "query_inventory",
                    "arguments": '{"sku_id": "SKU-998"}'
                }
            }
        else:
            return {
                "content": "根据库存系统查询结果，商品 SKU-998 当前库存剩余 42 件，存放于货位 A-03-12。"
            }

    agent = MinimalReActAgent(tools, max_turns=3)
    
    # 构造假图片字节
    fake_img = Image.new('RGB', (2048, 1536), color = 'red')
    img_byte_arr = io.BytesIO()
    fake_img.save(img_byte_arr, format='JPEG')
    
    final_output = asyncio.run(
        agent.run("帮我查一下这张图片里提及的 SKU-998 商品库存", img_byte_arr.getvalue(), mock_multimodal_llm)
    )
    print(f"\n[Agent 终态响应]\n{final_output}")

5. 从 MVP 到渐进式演进：何时引入 Multi-Agent

千万不要为了架构图好看而过早设计 Multi-Agent。

只有当单个 Agent 的 Tool 数量超过 15 个，导致提示词注意力和 Tool 选择准确率出现明显下滑；或者不同任务分支对系统权限、数据隔离（如财务数据与公开客服数据）有硬性要求时，才考虑把大 Agent 拆分为多个相互隔离的专用子 Agent。

自顶向下的复杂设计容易带来灾难，自底向上由 MVP 演进出的系统，才具备面向生产环境的生命力。

把环境条件和结果放在一起

这篇主题里，最值得先核实的不是概念是否漂亮，而是哪一步真的改变了结果。多模态原型先确定每种输入的缺失策略，图像失败、音频为空时系统该怎么退回必须明确。 把这一步单独拎出来观察，通常比同时调整一串参数更快找到问题。

我倾向于把异常样本保留下来：请求是什么、当时用了什么配置、返回内容或错误落在哪一层。正常样本只能说明流程曾经跑通，异常样本才会暴露接口假设、资源限制和交接位置。

如果需要扩大范围，也应先把原有行为放在旁边对照。新旧差异说得清楚，讨论才不会停留在感觉变快了或好像更稳定这种无法落地的判断上。

回到“AI Agent 系统设计与多模态交互实验：从最小可用方案搭起”，先把这些信号接到现有工作流。缺少必要信息时应明确标为待确认，不能用想象补上细节。