Agent多模态知识融合与推理

现实世界的信息天然是多模态的——一段医疗记录包含文字描述、CT影像、化验数值；一个软件缺陷报告包含错误日志、截图、堆栈信息；一个产品说明书包含文字说明、结构图、参数表格。仅处理文本的单模态Agent无法真正理解这些复杂场景。多模态知识融合与推理是Agent从"能读会写"进化到"能看会算"的关键能力，它让Agent能够同时理解文本、图像、代码、表格等不同模态的信息，并跨模态进行推理和决策。

一、多模态融合架构：从早期融合到晚期融合

多模态融合的核心挑战是：不同模态的数据具有不同的特征空间——文本是离散的符号序列，图像是连续的像素矩阵，代码是结构化的语法树。如何将这些异构信息统一到一个推理框架中，是多模态Agent的首要问题。

融合架构主要有三种范式：早期融合、晚期融合、混合融合。

早期融合在特征提取阶段就将不同模态的原始数据或低层特征合并。典型做法是将图像通过CNN提取特征图，将文本通过Transformer提取token序列，然后在模型的底层就进行跨模态注意力交互。早期融合的优点是模态间信息交互充分，能捕捉细粒度的跨模态关联；缺点是训练计算量大，且模态缺失时处理困难。

晚期融合让每个模态独立编码，在决策层才合并各模态的输出。比如图像分支输出"这是一张肺部CT"，文本分支输出"患者咳嗽发热"，决策层综合两者得出"可能为肺炎"。晚期融合的优点是模块化设计，模态可增可减，训练灵活；缺点是模态间深层交互不足，难以捕捉"图中阴影区域与文本描述的病灶位置对应"这类细粒度关联。

混合融合结合两者优点：在中间层进行部分跨模态交互，在决策层进行最终融合。这是当前主流多模态大模型（如GPT-4V、Gemini）采用的架构——视觉编码器提取图像特征后，通过投影层映射到与文本token相同的嵌入空间，然后在统一的Transformer中进行深层交互。

import math
from dataclasses import dataclass
from typing import Optional

@dataclass
class ModalityInput:
    modality: str  # text / image / code / table
    raw_data: str
    embedding: Optional[list] = None

class MultimodalEncoder:
    def __init__(self, text_dim=768, image_dim=1024, code_dim=768, unified_dim=768):
        self.text_dim = text_dim
        self.image_dim = image_dim
        self.code_dim = code_dim
        self.unified_dim = unified_dim
        self.text_encoder = TextEncoder(text_dim)
        self.image_encoder = ImageEncoder(image_dim)
        self.code_encoder = CodeEncoder(code_dim)
        self.projection_layers = {
            "text": self._init_projection(text_dim, unified_dim),
            "image": self._init_projection(image_dim, unified_dim),
            "code": self._init_projection(code_dim, unified_dim),
        }

    def _init_projection(self, in_dim, out_dim):
        return [[random.gauss(0, 0.02) for _ in range(out_dim)] for _ in range(in_dim)]

    def encode(self, inputs: list) -> list:
        encoded = []
        for inp in inputs:
            if inp.modality == "text":
                raw_emb = self.text_encoder.encode(inp.raw_data)
                projected = self._project(raw_emb, self.projection_layers["text"])
            elif inp.modality == "image":
                raw_emb = self.image_encoder.encode(inp.raw_data)
                projected = self._project(raw_emb, self.projection_layers["image"])
            elif inp.modality == "code":
                raw_emb = self.code_encoder.encode(inp.raw_data)
                projected = self._project(raw_emb, self.projection_layers["code"])
            else:
                continue
            encoded.append({"modality": inp.modality, "embedding": projected, "raw": inp.raw_data})
        return encoded

    def _project(self, embedding, projection_matrix):
        result = [0.0] * self.unified_dim
        for i, val in enumerate(embedding):
            if i < len(projection_matrix):
                for j in range(self.unified_dim):
                    result[j] += val * projection_matrix[i][j]
        return result

class TextEncoder:
    def __init__(self, dim=768):
        self.dim = dim

    def encode(self, text: str) -> list:
        vec = [0.0] * self.dim
        for i, ch in enumerate(text[:self.dim]):
            vec[i] = ord(ch) / 65536.0
        return vec

class ImageEncoder:
    def __init__(self, dim=1024):
        self.dim = dim

    def encode(self, image_path: str) -> list:
        vec = [0.0] * self.dim
        hash_val = hash(image_path)
        for i in range(self.dim):
            vec[i] = ((hash_val >> (i % 32)) & 0xFF) / 255.0
        return vec

class CodeEncoder:
    def __init__(self, dim=768):
        self.dim = dim

    def encode(self, code: str) -> list:
        vec = [0.0] * self.dim
        tokens = code.replace("(", " ( ").replace(")", " ) ").split()
        for i, token in enumerate(tokens[:self.dim]):
            vec[i] = hash(token) % 1000 / 1000.0
        return vec

import random

class CrossModalAttention:
    def __init__(self, dim=768, num_heads=8):
        self.dim = dim
        self.num_heads = num_heads
        self.head_dim = dim // num_heads
        self.query_weights = self._init_weights(dim, dim)
        self.key_weights = self._init_weights(dim, dim)
        self.value_weights = self._init_weights(dim, dim)

    def _init_weights(self, in_d, out_d):
        return [[random.gauss(0, 0.02) for _ in range(out_d)] for _ in range(in_d)]

    def attend(self, encoded_inputs: list) -> list:
        if len(encoded_inputs) < 2:
            return encoded_inputs
        # 每个模态的embedding作为query，与其他模态做cross-attention
        enhanced = []
        for i, current in enumerate(encoded_inputs):
            query = self._linear(current["embedding"], self.query_weights)
            # 收集其他模态作为key和value
            others = [e for j, e in enumerate(encoded_inputs) if j != i]
            if not others:
                enhanced.append(current)
                continue
            keys = [self._linear(o["embedding"], self.key_weights) for o in others]
            values = [self._linear(o["embedding"], self.value_weights) for o in others]
            # 计算注意力
            attended = self._multi_head_attention(query, keys, values)
            # 残差连接
            merged = [a + b for a, b in zip(current["embedding"], attended)]
            enhanced.append({"modality": current["modality"], "embedding": merged, "raw": current["raw"]})
        return enhanced

    def _linear(self, vec, weights):
        out_dim = len(weights[0])
        result = [0.0] * out_dim
        for i, val in enumerate(vec):
            if i < len(weights):
                for j in range(out_dim):
                    result[j] += val * weights[i][j]
        return result

    def _multi_head_attention(self, query, keys, values):
        result = [0.0] * self.dim
        for h in range(self.num_heads):
            start = h * self.head_dim
            end = start + self.head_dim
            q_h = query[start:end]
            scores = []
            for k in keys:
                k_h = k[start:end]
                score = sum(a * b for a, b in zip(q_h, k_h)) / math.sqrt(self.head_dim)
                scores.append(score)
            # softmax
            max_s = max(scores) if scores else 0
            exps = [math.exp(s - max_s) for s in scores]
            total = sum(exps)
            attn = [e / total for e in exps] if total else [1.0 / len(scores)] * len(scores)
            # 加权求和
            for v_idx, v in enumerate(values):
                v_h = v[start:end]
                for j in range(self.head_dim):
                    result[start + j] += attn[v_idx] * v_h[j]
        return result

这个多模态编码器的核心设计在于：每个模态有独立的编码器提取模态特有特征，通过投影层将不同模态映射到统一的嵌入空间，跨模态注意力机制让不同模态的特征在统一空间中交互增强。CrossModalAttention实现了多头注意力，让每个模态都能"看到"其他模态的信息。

二、跨模态检索：用一种模态查询另一种模态

跨模态检索是多模态Agent的核心能力之一——用文本查询图像、用图像查询代码、用代码查询文档。比如用户描述"一个红色按钮在页面右上角"，Agent需要从UI截图中找到对应元素；用户上传一张错误截图，Agent需要从代码库中找到相关代码段。

跨模态检索的技术基础是对比学习（Contrastive Learning）。通过训练，让语义相似的跨模态样本在嵌入空间中距离接近，不相似的远离。CLIP是这一方向的里程碑工作——它用4亿对图文对训练，让图像编码器和文本编码器在共享嵌入空间中对齐。

在Agent系统中，跨模态检索需要解决三个工程问题：多模态索引构建、跨模态查询执行、检索结果排序与融合。

多模态索引构建时，将知识库中的所有多模态内容（文档、图片、代码片段、表格）分别编码为向量，存入统一的向量索引。关键是为每个向量标注模态类型和原始内容，以便检索后还原上下文。

跨模态查询执行时，将用户的查询（可能是任意模态）编码为向量，在索引中检索最近邻。由于不同模态已经对齐到统一空间，文本查询可以直接检索到相关图像，图像查询可以直接检索到相关文本。

检索结果排序与融合时，需要综合考虑语义相似度、模态匹配度、上下文相关性。比如用户用文本查询时，返回的图像结果需要附带文本描述，否则Agent无法直接使用图像向量做推理。

from dataclasses import dataclass
from typing import Optional
import math

@dataclass
class MultimodalIndexEntry:
    entry_id: str
    modality: str
    content: str
    embedding: list
    description: str  # 跨模态描述，用于其他模态理解
    metadata: dict

class CrossModalRetriever:
    def __init__(self, encoder: MultimodalEncoder, top_k=5):
        self.encoder = encoder
        self.top_k = top_k
        self.index = []

    def add_entries(self, entries: list):
        for entry in entries:
            inp = ModalityInput(modality=entry.modality, raw_data=entry.content)
            encoded = self.encoder.encode([inp])
            if encoded:
                entry.embedding = encoded[0]["embedding"]
                self.index.append(entry)

    def search(self, query: str, query_modality: str = "text", target_modalities: list = None) -> list:
        query_inp = ModalityInput(modality=query_modality, raw_data=query)
        encoded = self.encoder.encode([query_inp])
        if not encoded:
            return []
        query_vec = encoded[0]["embedding"]
        # 计算与所有索引项的相似度
        scored = []
        for entry in self.index:
            if target_modalities and entry.modality not in target_modalities:
                continue
            sim = self._cosine_similarity(query_vec, entry.embedding)
            scored.append({"entry": entry, "similarity": sim})
        scored.sort(key=lambda x: x["similarity"], reverse=True)
        return scored[:self.top_k]

    def search_multimodal(self, query_inputs: list) -> list:
        """多模态联合查询"""
        all_results = []
        for inp in query_inputs:
            results = self.search(inp.raw_data, inp.modality)
            all_results.extend(results)
        # 去重并合并分数
        merged = {}
        for r in all_results:
            eid = r["entry"].entry_id
            if eid in merged:
                merged[eid]["similarity"] = max(merged[eid]["similarity"], r["similarity"])
            else:
                merged[eid] = r
        final = list(merged.values())
        final.sort(key=lambda x: x["similarity"], reverse=True)
        return final[:self.top_k]

    def build_context(self, results: list) -> str:
        sections = []
        for r in results:
            entry = r["entry"]
            sections.append(f"[{entry.modality}] (相似度: {r['similarity']:.2f})")
            sections.append(f"  内容: {entry.content[:200]}")
            sections.append(f"  描述: {entry.description}")
            if entry.metadata:
                sections.append(f"  元数据: {entry.metadata}")
        return "\n".join(sections)

    def _cosine_similarity(self, vec_a, vec_b):
        dot = sum(a * b for a, b in zip(vec_a, vec_b))
        norm_a = math.sqrt(sum(a * a for a in vec_a))
        norm_b = math.sqrt(sum(b * b for b in vec_b))
        return dot / (norm_a * norm_b) if norm_a and norm_b else 0.0

这个跨模态检索器的核心在于：统一索引支持所有模态的存储和检索；search_multimodal方法支持多模态联合查询，比如同时用文本和图像检索相关内容；build_context方法将检索结果格式化为Agent可理解的上下文，包含模态类型、相似度、原始内容和跨模态描述。

三、多模态推理链：跨模态的逐步推理

单模态推理链是纯文本的"因为A所以B"链条。多模态推理链则涉及跨模态的推理步骤——从图像中观察到现象，结合文本知识得出假设，通过代码验证假设，最终给出结论。

比如医疗诊断场景的多模态推理链：第一步，从CT影像中识别出肺部阴影（图像模态）；第二步，结合患者症状描述"咳嗽发热两周"（文本模态），推断可能为感染；第三步，查看化验结果"白细胞升高"（表格模态），支持细菌感染判断；第四步，综合以上证据，诊断为细菌性肺炎，建议抗生素治疗。

多模态推理链的构建需要三个能力：模态内推理（在单个模态内做逻辑推理）、跨模态关联（将不同模态的信息关联起来）、推理链组装（将多步推理组织成连贯链条）。

模态内推理依赖各模态的专用推理器——文本推理器做文本逻辑推导，视觉推理器做空间关系判断，代码推理器做执行验证。跨模态关联依赖前述的跨模态注意力机制，将不同模态的中间表示对齐。推理链组装需要一个控制器，决定推理顺序、何时切换模态、何时终止推理。

from dataclasses import dataclass, field
from typing import Optional
from enum import Enum

class ReasoningStepType(Enum):
    TEXT = "text"
    VISUAL = "visual"
    CODE = "code"
    TABLE = "table"
    CROSS_MODAL = "cross_modal"

@dataclass
class ReasoningStep:
    step_id: int
    step_type: ReasoningStepType
    input_modalities: list
    reasoning: str
    conclusion: str
    evidence: list  # 引用的证据片段
    confidence: float

@dataclass
class MultimodalReasoningChain:
    chain_id: str
    query: str
    steps: list = field(default_factory=list)
    final_conclusion: str = ""
    overall_confidence: float = 0.0

class MultimodalReasoner:
    def __init__(self, encoder: MultimodalEncoder, retriever: CrossModalRetriever):
        self.encoder = encoder
        self.retriever = retriever
        self.max_steps = 10

    def reason(self, query: str, available_inputs: list) -> MultimodalReasoningChain:
        chain = MultimodalReasoningChain(chain_id=f"chain_{hash(query)}", query=query)
        # 第一步：理解查询，检索相关知识
        step1 = self._understand_query(query, available_inputs)
        chain.steps.append(step1)
        # 后续步骤：逐步推理
        current_conclusion = step1.conclusion
        for i in range(1, self.max_steps):
            # 判断是否需要继续推理
            if self._is_sufficient(current_conclusion, chain.steps):
                break
            # 决定下一步推理类型
            next_step = self._next_step(chain.steps, available_inputs, query)
            if next_step is None:
                break
            chain.steps.append(next_step)
            current_conclusion = next_step.conclusion
        # 组装最终结论
        chain.final_conclusion = self._synthesize(chain.steps)
        chain.overall_confidence = self._aggregate_confidence(chain.steps)
        return chain

    def _understand_query(self, query: str, inputs: list) -> ReasoningStep:
        # 检索相关多模态知识
        results = self.retriever.search(query, "text")
        context = self.retriever.build_context(results)
        reasoning = f"分析用户查询: {query}\n检索到相关多模态知识:\n{context[:500]}"
        conclusion = f"查询涉及以下模态信息: {', '.join(set(r['entry'].modality for r in results))}"
        evidence = [r["entry"].content[:100] for r in results[:3]]
        return ReasoningStep(step_id=0, step_type=ReasoningStepType.CROSS_MODAL,
            input_modalities=["text"], reasoning=reasoning, conclusion=conclusion,
            evidence=evidence, confidence=0.7)

    def _next_step(self, previous_steps: list, available_inputs: list, query: str) -> Optional[ReasoningStep]:
        last_step = previous_steps[-1]
        # 检查是否有未使用的模态输入
        used_modalities = set()
        for step in previous_steps:
            used_modalities.update(step.input_modalities)
        unused = [inp for inp in available_inputs if inp.modality not in used_modalities]
        if unused:
            # 对未使用的模态进行推理
            inp = unused[0]
            encoded = self.encoder.encode([inp])
            if encoded:
                reasoning = f"分析{inp.modality}模态输入: {inp.raw_data[:200]}"
                conclusion = f"从{inp.modality}模态获得信息: {inp.raw_data[:100]}"
                return ReasoningStep(step_id=len(previous_steps), step_type=ReasoningStepType(inp.modality),
                    input_modalities=[inp.modality], reasoning=reasoning, conclusion=conclusion,
                    evidence=[inp.raw_data[:100]], confidence=0.6)
        # 跨模态综合推理
        all_conclusions = " ".join([s.conclusion for s in previous_steps[-3:]])
        reasoning = f"综合前几步结论进行跨模态推理: {all_conclusions[:300]}"
        conclusion = f"综合推理: 基于多模态证据，得出阶段性结论"
        return ReasoningStep(step_id=len(previous_steps), step_type=ReasoningStepType.CROSS_MODAL,
            input_modalities=[s.step_type.value for s in previous_steps[-3:]],
            reasoning=reasoning, conclusion=conclusion, evidence=[], confidence=0.65)

    def _is_sufficient(self, conclusion: str, steps: list) -> bool:
        if len(steps) >= self.max_steps:
            return True
        if len(steps) >= 3 and any("最终" in s.conclusion or "结论" in s.conclusion for s in steps[-2:]):
            return True
        return False

    def _synthesize(self, steps: list) -> str:
        conclusions = [f"步骤{s.step_id}: {s.conclusion}" for s in steps]
        return " -> ".join(conclusions)

    def _aggregate_confidence(self, steps: list) -> float:
        if not steps:
            return 0.0
        # 置信度随步骤增加而衰减
        total = 0.0
        weight_sum = 0.0
        for i, step in enumerate(steps):
            weight = 0.8 ** i
            total += step.confidence * weight
            weight_sum += weight
        return total / weight_sum if weight_sum else 0.0

    def format_chain(self, chain: MultimodalReasoningChain) -> str:
        lines = [f"查询: {chain.query}", f"整体置信度: {chain.overall_confidence:.2f}", "推理链:"]
        for step in chain.steps:
            lines.append(f"  [{step.step_id}] 类型:{step.step_type.value} 置信度:{step.confidence:.2f}")
            lines.append(f"      推理: {step.reasoning[:150]}")
            lines.append(f"      结论: {step.conclusion}")
        lines.append(f"最终结论: {chain.final_conclusion}")
        return "\n".join(lines)

这个多模态推理器的核心设计在于：推理链由多个步骤组成，每步标注模态类型和置信度；_next_step方法根据当前推理状态决定下一步的模态和推理类型；置信度随推理深度衰减，反映推理链越长不确定性越高的特性；format_chain方法将推理链格式化为可读文本，支持Agent将推理过程展示给用户。

四、模态对齐：让不同模态说同一种语言

模态对齐是多模态融合的基础——只有当不同模态的表示在同一个语义空间中对齐，跨模态检索和推理才能有效工作。模态对齐的训练方法主要有三种：对比学习对齐、生成式对齐、共享编码器对齐。

对比学习对齐以CLIP为代表，用大量配对数据（如图文对）训练，让配对的跨模态样本在嵌入空间中接近，不配对的远离。训练目标是最大化配对样本的余弦相似度，最小化非配对样本的相似度。对比学习的优点是不需要标注数据，可以用大规模无标注配对数据训练；缺点是需要海量数据才能达到好的对齐效果。

生成式对齐通过跨模态生成任务实现——比如图像描述生成（看图说话）让模型学会从图像模态映射到文本模态，文本生成图像让模型学会反向映射。生成式对齐的优点是对齐关系更细粒度；缺点是训练成本高，且生成质量影响对齐效果。

共享编码器对齐使用统一的编码器处理所有模态——比如将图像分割为patch序列，用与文本相同的Transformer编码器处理。共享编码器天然实现模态对齐，因为所有模态使用相同的参数空间。ViT（Vision Transformer）就是这一思路的代表。

在Agent系统中，模态对齐质量直接影响多模态推理的准确性。对齐不好会导致跨模态检索召回率低、跨模态推理出现"张冠李戴"的错误。

import random
from dataclasses import dataclass

@dataclass
class AlignmentPair:
    source_modality: str
    source_data: str
    target_modality: str
    target_data: str
    label: int  # 1=positive pair, 0=negative pair

class ContrastiveAlignmentTrainer:
    def __init__(self, encoder: MultimodalEncoder, temperature=0.07, learning_rate=1e-4):
        self.encoder = encoder
        self.temperature = temperature
        self.learning_rate = learning_rate
        self.training_log = []

    def train_step(self, pairs: list) -> dict:
        positive_pairs = [p for p in pairs if p.label == 1]
        negative_pairs = [p for p in pairs if p.label == 0]
        if not positive_pairs:
            return {"loss": 0.0, "positive_count": 0}
        total_loss = 0.0
        for pos in positive_pairs:
            src_inp = ModalityInput(modality=pos.source_modality, raw_data=pos.source_data)
            tgt_inp = ModalityInput(modality=pos.target_modality, raw_data=pos.target_data)
            encoded = self.encoder.encode([src_inp, tgt_inp])
            if len(encoded) < 2:
                continue
            src_vec = encoded[0]["embedding"]
            tgt_vec = encoded[1]["embedding"]
            # 正样本相似度
            pos_sim = self._cosine_sim(src_vec, tgt_vec) / self.temperature
            # 负样本相似度
            neg_sims = []
            for neg in negative_pairs[:10]:
                neg_inp = ModalityInput(modality=neg.target_modality, raw_data=neg.target_data)
                neg_encoded = self.encoder.encode([neg_inp])
                if neg_encoded:
                    neg_sims.append(self._cosine_sim(src_vec, neg_encoded[0]["embedding"]) / self.temperature)
            # InfoNCE损失
            if neg_sims:
                all_sims = [pos_sim] + neg_sims
                max_sim = max(all_sims)
                exps = [math.exp(s - max_sim) for s in all_sims]
                loss = -math.log(exps[0] / sum(exps))
                total_loss += loss
                # 简化梯度更新：拉近正样本、推远负样本
                self._update_alignment(src_vec, tgt_vec, neg_sims, neg_encoded if neg_sims else None)
        avg_loss = total_loss / len(positive_pairs)
        self.training_log.append({"loss": avg_loss, "pairs": len(positive_pairs)})
        return {"loss": avg_loss, "positive_count": len(positive_pairs), "negative_count": len(negative_pairs)}

    def _update_alignment(self, src_vec, tgt_vec, neg_sims, neg_encoded):
        # 拉近正样本对
        for i in range(len(src_vec)):
            if i < len(tgt_vec):
                diff = src_vec[i] - tgt_vec[i]
                # 简化的梯度方向
                pass  # 实际中更新投影层参数

    def evaluate_alignment(self, test_pairs: list) -> dict:
        positive_sims = []
        negative_sims = []
        for pair in test_pairs:
            src_inp = ModalityInput(modality=pair.source_modality, raw_data=pair.source_data)
            tgt_inp = ModalityInput(modality=pair.target_modality, raw_data=pair.target_data)
            encoded = self.encoder.encode([src_inp, tgt_inp])
            if len(encoded) >= 2:
                sim = self._cosine_sim(encoded[0]["embedding"], encoded[1]["embedding"])
                if pair.label == 1:
                    positive_sims.append(sim)
                else:
                    negative_sims.append(sim)
        avg_pos = sum(positive_sims) / len(positive_sims) if positive_sims else 0
        avg_neg = sum(negative_sims) / len(negative_sims) if negative_sims else 0
        # 计算对齐质量指标
        gap = avg_pos - avg_neg
        return {"avg_positive_sim": avg_pos, "avg_negative_sim": avg_neg,
                "alignment_gap": gap, "alignment_quality": "good" if gap > 0.2 else "needs_improvement"}

    def _cosine_sim(self, vec_a, vec_b):
        dot = sum(a * b for a, b in zip(vec_a, vec_b))
        norm_a = math.sqrt(sum(a * a for a in vec_a))
        norm_b = math.sqrt(sum(b * b for b in vec_b))
        return dot / (norm_a * norm_b) if norm_a and norm_b else 0.0

这个对齐训练器的核心在于：使用InfoNCE对比损失函数，拉近语义配对的跨模态样本、推远非配对样本；evaluate_alignment方法量化评估对齐质量，正负样本相似度差距越大对齐越好。

五、完整多模态Agent：整合所有能力

将多模态编码、跨模态检索、多模态推理、模态对齐整合为一个完整的多模态Agent，需要建立从多模态输入到多模态输出的完整链路。

class MultimodalAgent:
    def __init__(self, encoder: MultimodalEncoder, retriever: CrossModalRetriever,
                 reasoner: MultimodalReasoner, base_model=None):
        self.encoder = encoder
        self.retriever = retriever
        self.reasoner = reasoner
        self.base_model = base_model
        self.conversation_history = []

    def process(self, user_inputs: list, query: str = "") -> str:
        # 编码所有输入模态
        encoded = self.encoder.encode(user_inputs)
        # 跨模态注意力增强
        attention = CrossModalAttention(dim=self.encoder.unified_dim)
        enhanced = attention.attend(encoded)
        # 检索相关知识
        search_inputs = [ModalityInput(modality=e["modality"], raw_data=e["raw"]) for e in enhanced]
        retrieved = self.retriever.search_multimodal(search_inputs)
        context = self.retriever.build_context(retrieved)
        # 多模态推理
        if query:
            chain = self.reasoner.reason(query, user_inputs)
            reasoning_text = self.reasoner.format_chain(chain)
        else:
            query = user_inputs[0].raw_data if user_inputs else ""
            chain = self.reasoner.reason(query, user_inputs)
            reasoning_text = self.reasoner.format_chain(chain)
        # 生成最终回答
        prompt = self._build_prompt(query, context, reasoning_text, enhanced)
        if self.base_model:
            response = self.base_model.generate(prompt)
        else:
            response = self._template_response(query, context, reasoning_text)
        # 记录对话历史
        self.conversation_history.append({
            "inputs": [{"modality": i.modality, "data": i.raw_data[:200]} for i in user_inputs],
            "query": query, "response": response[:500]
        })
        return response

    def _build_prompt(self, query, context, reasoning, enhanced):
        modality_summary = ", ".join([f"{e['modality']}" for e in enhanced])
        return (f"多模态输入模态: {modality_summary}\n\n"
                f"检索到的多模态知识:\n{context[:1000]}\n\n"
                f"多模态推理链:\n{reasoning[:1000]}\n\n"
                f"用户问题: {query}\n\n"
                f"请基于以上多模态信息给出综合回答:")

    def _template_response(self, query, context, reasoning):
        lines = [f"基于多模态分析，对查询「{query}」的回答:", "", "推理过程:"]
        lines.append(reasoning[:500])
        lines.append("")
        lines.append("参考知识:")
        lines.append(context[:500])
        return "\n".join(lines)

    def add_knowledge(self, modality: str, content: str, description: str, metadata: dict = None):
        entry = MultimodalIndexEntry(entry_id=f"entry_{len(self.retriever.index)}",
            modality=modality, content=content, embedding=[], description=description,
            metadata=metadata or {})
        self.retriever.add_entries([entry])

    def get_capabilities(self) -> dict:
        return {
            "supported_modalities": ["text", "image", "code"],
            "indexed_entries": len(self.retriever.index),
            "modality_distribution": self._modality_stats(),
            "conversation_count": len(self.conversation_history)
        }

    def _modality_stats(self) -> dict:
        stats = {}
        for entry in self.retriever.index:
            stats[entry.modality] = stats.get(entry.modality, 0) + 1
        return stats

这个完整多模态Agent的工作流程是：接收多模态输入后，先用编码器将各模态编码到统一空间，通过跨模态注意力增强模态间信息交互，检索多模态知识库获取相关上下文，执行多模态推理链逐步推导结论，最终生成综合回答。add_knowledge方法支持向知识库动态添加多模态内容，get_capabilities方法报告Agent当前的多模态能力概况。

从产业实践来看，多模态Agent在智能客服（理解用户截图和文字描述）、医疗辅助（结合影像和病历）、工业质检（分析产品图片和规格参数）、代码审查（理解代码和UI截图）等场景中展现出巨大潜力。但多模态Agent的落地也面临挑战：多模态数据的获取和标注成本高、模态对齐质量在长尾分布上不稳定、多模态推理链的可解释性难以保证、不同模态的信息密度差异大导致推理不平衡。

未来随着多模态大模型能力的提升和多模态数据生态的完善，多模态Agent将从当前的"文本为主、其他模态辅助"模式进化到"多模态平等协作"模式，真正实现像人类一样自然地处理多感官信息。