多模态应用架构中的需求澄清机制设计与工程权衡

multimodal-clarification-engine/
├── app/
│   ├── __init__.py
│   ├── main.py              # FastAPI 应用入口
│   ├── core/
│   │   ├── __init__.py
│   │   ├── engine.py        # 核心澄清引擎
│   │   ├── decision_maker.py # 澄清决策器
│   │   └── config.py        # 配置管理
│   ├── processors/
│   │   ├── __init__.py
│   │   ├── text_processor.py
│   │   └── vision_processor.py # 图像处理与多模态模型调用
│   ├── schemas/
│   │   ├── __init__.py
│   │   └── models.py        # Pydantic 数据模型
│   └── utils/
│       ├── __init__.py
│       └── clarifier.py     # 澄清问题生成工具
├── requirements.txt
├── config.yaml              # 应用配置文件
└── run.py                   # 启动脚本

engine:
  clarification_threshold: 0.65 # 确定性低于此阈值则触发澄清
  max_clarification_options: 4  # 澄清问题中最多提供的选项数

models:
  vit_model: "google/vit-base-patch16-224"
  blip_model: "Salesforce/blip-image-captioning-base"
  local_mode: true # 为true时使用transformers本地模型，false可配置为远程API

server:
  host: "0.0.0.0"
  port: 8000
  log_level: "info"

from pydantic import BaseModel, HttpUrl
from typing import List, Optional, Any
from enum import Enum

class ModalityType(str, Enum):
    TEXT = "text"
    IMAGE = "image"
    MULTIMODAL = "multimodal"

class UserRequest(BaseModel):
    """用户请求数据模型"""
    request_id: str
    text_input: Optional[str] = None
    image_url: Optional[HttpUrl] = None
    # 实际部署中，可能使用 base64 或文件上传，这里用 URL 简化演示
    modality: ModalityType

class ModelOutput(BaseModel):
    """单个模型输出的统一抽象"""
    content: Any  # 可以是字符串、列表、字典等
    confidence: float  # 模型自身置信度或我们计算出的确定性分数
    raw_response: Optional[Any] = None

class ClarificationOption(BaseModel):
    """一个澄清选项"""
    option_id: str
    description: str
    confidence_if_chosen: float = 1.0 # 如果用户选择此项，后续流程的置信度

class ClarificationQuestion(BaseModel):
    """生成的澄清问题"""
    question_id: str
    question_text: str
    options: List[ClarificationOption]

class EngineResponse(BaseModel):
    """引擎的最终响应"""
    request_id: str
    needs_clarification: bool
    clarification: Optional[ClarificationQuestion] = None
    immediate_action: Optional[dict] = None  # 如果不需要澄清，直接执行的动作
    processed_results: dict  # 各处理器输出的中间结果，用于调试
    overall_confidence: float

import torch
from PIL import Image
import requests
from io import BytesIO
from transformers import BlipProcessor, BlipForConditionalGeneration, ViTForImageClassification, ViTImageProcessor
import logging
from typing import Tuple, Dict, Any
from app.core.config import get_settings

settings = get_settings()
logger = logging.getLogger(__name__)

class VisionProcessor:
    """处理图像模态，集成视觉问答和图像描述能力"""

    def __init__(self):
        self.device = "cuda" if torch.cuda.is_available() else "cpu"
        logger.info(f"Initializing VisionProcessor on device: {self.device}")
        self._load_models()

    def _load_models(self):
        """加载多模态模型。本地模式加载本地模型，否则可初始化API客户端"""
        if settings.models.local_mode:
            # 1. 图像描述模型 (BLIP)
            self.blip_processor = BlipProcessor.from_pretrained(settings.models.blip_model)
            self.blip_model = BlipForConditionalGeneration.from_pretrained(settings.models.blip_model).to(self.device)
            # 2. 图像分类模型 (ViT) - 用于获取场景标签/对象信息
            self.vit_processor = ViTImageProcessor.from_pretrained(settings.models.vit_model)
            self.vit_model = ViTForImageClassification.from_pretrained(settings.models.vit_model).to(self.device)
        else:
            # 此处可替换为调用云端视觉API（如GPT-4V, Claude-3）的客户端
            self.blip_model = None
            self.vit_model = None
            logger.warning("Remote API mode not fully implemented, using stub functions.")

    def _download_image(self, image_url: str) -> Image.Image:
        """从URL下载图像"""
        try:
            response = requests.get(image_url, timeout=10)
            response.raise_for_status()
            image = Image.open(BytesIO(response.content)).convert("RGB")
            return image
        except Exception as e:
            logger.error(f"Failed to download image from {image_url}: {e}")
            raise

    def describe_image(self, image_url: str) -> ModelOutput:
        """生成图像的文本描述"""
        try:
            image = self._download_image(image_url)
            if settings.models.local_mode:
                inputs = self.blip_processor(image, return_tensors="pt").to(self.device)
                with torch.no_grad():
                    out = self.blip_model.generate(**inputs, max_new_tokens=50)
                description = self.blip_processor.decode(out[0], skip_special_tokens=True)
                # BLIP不直接输出置信度，这里用一个启发式分数（例如，生成文本的softmax概率均值，简化用固定值）
                confidence = 0.85
            else:
                # 模拟远程API调用
                description = f"A description of image from {image_url}"
                confidence = 0.80
            return ModelOutput(content=description, confidence=confidence, raw_response=description)
        except Exception as e:
            logger.error(f"Image description failed: {e}")
            return ModelOutput(content="", confidence=0.0, raw_response=str(e))

    def detect_objects_and_scene(self, image_url: str) -> ModelOutput:
        """检测图像中的主要对象和场景，返回标签和置信度"""
        try:
            image = self._download_image(image_url)
            if settings.models.local_mode:
                inputs = self.vit_processor(images=image, return_tensors="pt").to(self.device)
                with torch.no_grad():
                    outputs = self.vit_model(**inputs)
                logits = outputs.logits
                probs = torch.nn.functional.softmax(logits, dim=-1)
                top_probs, top_indices = torch.topk(probs, 5) # 取前5个可能类别
                labels = [self.vit_model.config.id2label[idx.item()] for idx in top_indices[0]]
                confidences = top_probs[0].tolist()
                result = {"labels": labels, "confidences": confidences}
                # 整体置信度用最高概率表示
                overall_confidence = confidences[0]
            else:
                result = {"labels": ["furniture", "living room"], "confidences": [0.9, 0.85]}
                overall_confidence = 0.9
            return ModelOutput(content=result, confidence=overall_confidence, raw_response=result)
        except Exception as e:
            logger.error(f"Object detection failed: {e}")
            return ModelOutput(content={"labels":[], "confidences":[]}, confidence=0.0, raw_response=str(e))

    def process(self, image_url: str) -> Dict[str, ModelOutput]:
        """处理图像URL，返回多种分析结果"""
        results = {
            "description": self.describe_image(image_url),
            "detection": self.detect_objects_and_scene(image_url)
        }
        return results

import logging
from typing import List, Dict, Any
from app.schemas.models import ModelOutput
from app.core.config import get_settings
import re

settings = get_settings()
logger = logging.getLogger(__name__)

class TextProcessor:
    """处理文本模态，进行意图分类和关键实体提取（简化版）"""

    def __init__(self):
        # 在实际项目中，这里会加载BERT/NLU模型。此处为演示，使用规则和简单逻辑。
        self.intent_keywords = {
            "recommendation": ["推荐", "买什么", "哪个好", "suggest", "recommend"],
            "comparison": ["对比", "比较", "vs", "versus", "compare"],
            "identification": ["这是什么", "是什么", "识别", "what is", "identify"]
        }
        self.ambiguous_pronouns = ["这个", "那个", "它", "它们", "这个", "that", "it", "them"]

    def analyze_intent(self, text: str) -> ModelOutput:
        """分析文本意图，返回意图类别和置信度"""
        if not text:
            return ModelOutput(content="unknown", confidence=0.0)
        text_lower = text.lower()
        scores = {}
        for intent, keywords in self.intent_keywords.items():
            score = sum([1 for kw in keywords if kw in text_lower])
            scores[intent] = score / len(keywords) if keywords else 0
        # 找到得分最高的意图
        if scores:
            best_intent = max(scores, key=scores.get)
            best_score = scores[best_intent]
            # 归一化到0-1区间
            confidence = min(best_score * 2, 1.0) # 简单缩放
        else:
            best_intent = "unknown"
            confidence = 0.3
        return ModelOutput(content=best_intent, confidence=confidence)

    def extract_entities_and_check_ambiguity(self, text: str) -> ModelOutput:
        """提取实体并检查指代模糊性"""
        entities = []
        is_ambiguous = False
        # 简单规则：检查是否存在模糊代词
        for pronoun in self.ambiguous_pronouns:
            if pronoun in text:
                is_ambiguous = True
                break
        # 简单名词短语提取（通过空格和常见标点分割）
        words = re.findall(r'[\w\u4e00-\u9fff]+', text) # 匹配中英文单词
        # 假设长度大于1的词为潜在实体（非常简化）
        entities = [word for word in words if len(word) > 1]
        result = {
            "entities": entities,
            "is_ambiguous": is_ambiguous,
            "ambiguous_pronouns_found": is_ambiguous
        }
        # 模糊性越高，置信度越低
        confidence = 0.9 if not is_ambiguous else 0.4
        return ModelOutput(content=result, confidence=confidence)

    def process(self, text: str) -> Dict[str, ModelOutput]:
        """处理文本，返回分析结果"""
        results = {
            "intent": self.analyze_intent(text),
            "entity_analysis": self.extract_entities_and_check_ambiguity(text)
        }
        return results

from typing import Dict
from app.schemas.models import ModelOutput
from app.core.config import get_settings
import logging

settings = get_settings()
logger = logging.getLogger(__name__)

class DecisionMaker:
    """
    澄清决策器。
    核心功能：综合多模态处理结果，计算整体确定性分数，并与阈值比较，决定是否触发澄清。
    """

    def __init__(self, threshold: float = None):
        self.threshold = threshold or settings.engine.clarification_threshold

    def compute_overall_confidence(self, processed_results: Dict[str, Dict[str, ModelOutput]]) -> float:
        """
        基于多模态处理结果计算整体置信度。
        这是一个加权融合策略，可以根据业务逻辑调整。
        """
        weights = {
            "text_intent": 0.3,
            "text_ambiguity": 0.4, # 文本模糊性权重较高
            "vision_description": 0.15,
            "vision_detection": 0.15
        }
        weighted_sum = 0.0
        total_weight = 0.0

        # 从处理结果中提取关键置信度
        text_results = processed_results.get("text", {})
        vision_results = processed_results.get("vision", {})

        if text_results:
            intent_output = text_results.get("intent")
            entity_output = text_results.get("entity_analysis")
            if intent_output:
                weighted_sum += weights["text_intent"] * intent_output.confidence
                total_weight += weights["text_intent"]
            if entity_output:
                # 文本模糊性高会拉低整体置信度
                weighted_sum += weights["text_ambiguity"] * entity_output.confidence
                total_weight += weights["text_ambiguity"]

        if vision_results:
            desc_output = vision_results.get("description")
            det_output = vision_results.get("detection")
            if desc_output:
                weighted_sum += weights["vision_description"] * desc_output.confidence
                total_weight += weights["vision_description"]
            if det_output:
                weighted_sum += weights["vision_detection"] * det_output.confidence
                total_weight += weights["vision_detection"]

        # 防止除零
        if total_weight == 0:
            return 0.0
        overall_confidence = weighted_sum / total_weight
        logger.debug(f"Computed overall confidence: {overall_confidence:.3f}")
        return overall_confidence

    def decide(self, processed_results: Dict) -> tuple[bool, float]:
        """做出澄清决策"""
        overall_confidence = self.compute_overall_confidence(processed_results)
        needs_clarification = overall_confidence < self.threshold
        logger.info(f"Decision: needs_clarification={needs_clarification}, confidence={overall_confidence:.3f}, threshold={self.threshold}")
        return needs_clarification, overall_confidence

from typing import List, Dict
from app.schemas.models import ClarificationQuestion, ClarificationOption
import uuid

class ClarificationGenerator:
    """根据处理结果生成澄清问题"""

    @staticmethod
    def generate_for_ambiguous_entity(
        vision_detection_result: Dict,
        text_entities: List[str],
        max_options: int
    ) -> ClarificationQuestion:
        """
        针对模糊指代（如"这个"）生成澄清问题。
        结合视觉检测到的对象和文本中提到的实体，生成候选选项。
        """
        # 从视觉结果中获取候选对象标签
        candidate_labels = vision_detection_result.get("labels", [])[:max_options]
        # 也可以结合文本实体，这里简单处理
        all_candidates = list(set(candidate_labels + text_entities))
        all_candidates = all_candidates[:max_options]

        if not all_candidates:
            all_candidates = ["图中的主要物体", "背景", "颜色或风格"]

        options = []
        for i, candidate in enumerate(all_candidates):
            option_id = f"opt_{i+1}"
            options.append(
                ClarificationOption(
                    option_id=option_id,
                    description=f""{candidate}"",
                    confidence_if_chosen=0.95 # 用户选择后，置信度提升
                )
            )
        # 添加一个"以上都不是"的选项
        options.append(
            ClarificationOption(
                option_id="opt_none",
                description="以上都不是，是其他东西",
                confidence_if_chosen=0.5
            )
        )

        question = ClarificationQuestion(
            question_id=str(uuid.uuid4())[:8],
            question_text="您刚才提到的"这个"，具体指的是图中的哪个部分？",
            options=options
        )
        return question

    @staticmethod
    def generate_for_intent_disambiguation(intent_scores: Dict) -> ClarificationQuestion:
        """当意图不明确时，生成澄清问题"""
        # 简化实现
        options = []
        possible_intents = list(intent_scores.keys())[:3]
        for i, intent in enumerate(possible_intents):
            intent_desc = {
                "recommendation": "希望我为您推荐商品",
                "comparison": "希望我对比不同商品",
                "identification": "希望我识别图中的物体"
            }.get(intent, intent)
            options.append(
                ClarificationOption(
                    option_id=f"intent_{i+1}",
                    description=intent_desc
                )
            )
        question = ClarificationQuestion(
            question_id=str(uuid.uuid4())[:8],
            question_text="请问您的主要需求是什么？",
            options=options
        )
        return question

import logging
from typing import Dict, Any
from app.schemas.models import *
from app.processors.text_processor import TextProcessor
from app.processors.vision_processor import VisionProcessor
from app.core.decision_maker import DecisionMaker
from app.utils.clarifier import ClarificationGenerator
from app.core.config import get_settings

settings = get_settings()
logger = logging.getLogger(__name__)

class MultimodalClarificationEngine:
    """多模态澄清引擎，协调整个处理流程"""

    def __init__(self):
        self.text_processor = TextProcessor()
        self.vision_processor = VisionProcessor()
        self.decision_maker = DecisionMaker()
        self.clarifier = ClarificationGenerator()
        logger.info("MultimodalClarificationEngine initialized.")

    async def process_request(self, request: UserRequest) -> EngineResponse:
        """
        处理用户请求的核心工作流。

        1. 多模态解析
        2. 不确定性评估与决策
        3. 生成澄清或执行动作
        """
        logger.info(f"Processing request {request.request_id}, modality: {request.modality}")

        # Step 1: 多模态处理
        processed_results = {}
        if request.text_input:
            processed_results["text"] = self.text_processor.process(request.text_input)
        if request.image_url:
            processed_results["vision"] = self.vision_processor.process(str(request.image_url))

        # Step 2: 决策
        needs_clarification, overall_confidence = self.decision_maker.decide(processed_results)

        # Step 3: 生成响应
        clarification = None
        immediate_action = None

        if needs_clarification:
            # 根据分析结果生成针对性的澄清问题
            # 策略：优先处理模糊指代，其次处理意图不明
            text_analysis = processed_results.get("text", {}).get("entity_analysis", ModelOutput(content={}))
            vision_detection = processed_results.get("vision", {}).get("detection", ModelOutput(content={}))
            if (text_analysis.content.get("is_ambiguous") and
                vision_detection.content):
                # 模糊指代澄清
                text_entities = text_analysis.content.get("entities", [])
                clarification = self.clarifier.generate_for_ambiguous_entity(
                    vision_detection_result=vision_detection.content,
                    text_entities=text_entities,
                    max_options=settings.engine.max_clarification_options
                )
            else:
                # 意图澄清
                intent_output = processed_results.get("text", {}).get("intent")
                if intent_output and intent_output.confidence < 0.7:
                    # 简化：模拟意图分数
                    intent_scores = {"recommendation": 0.5, "comparison": 0.3}
                    clarification = self.clarifier.generate_for_intent_disambiguation(intent_scores)
                else:
                    # 兜底澄清问题
                    clarification = ClarificationQuestion(
                        question_id="gen_01",
                        question_text="您能更详细地描述一下您的需求吗？",
                        options=[ClarificationOption(option_id="detail", description="提供更多细节")]
                    )
        else:
            # 置信度高，执行预设业务逻辑（此处为模拟）
            intent = processed_results.get("text", {}).get("intent", ModelOutput(content="unknown"))
            immediate_action = {
                "action": "execute_recommendation",
                "target": intent.content,
                "message": f"基于您的请求和图片，已为您执行 {intent.content} 操作。"
            }

        # 构建最终响应
        response = EngineResponse(
            request_id=request.request_id,
            needs_clarification=needs_clarification,
            clarification=clarification,
            immediate_action=immediate_action,
            processed_results={k: {sk: so.dict() for sk, so in v.items()} for k, v in processed_results.items()}, # 转换为可序列化字典
            overall_confidence=overall_confidence
        )
        logger.info(f"Request {request.request_id} processed. Needs clarification: {needs_clarification}")
        return response

from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
import logging
import uuid
from app.schemas.models import UserRequest, EngineResponse
from app.core.engine import MultimodalClarificationEngine
from app.core.config import get_settings

settings = get_settings()

# 配置日志
logging.basicConfig(level=getattr(logging, settings.server.log_level.upper()))
logger = logging.getLogger(__name__)

app = FastAPI(title="Multimodal Clarification Engine API")

# 添加CORS中间件（便于前端测试）
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 全局引擎实例
engine = MultimodalClarificationEngine()

@app.get("/")
async def root():
    return {"message": "Multimodal Clarification Engine is running."}

@app.post("/api/clarify", response_model=EngineResponse)
async def clarify_request(request: UserRequest):
    """
    主API端点：接收多模态用户请求，返回是否需要澄清及相应内容。
    """
    try:
        # 为请求生成唯一ID（如果客户端未提供）
        if not request.request_id:
            request.request_id = str(uuid.uuid4())[:8]
        logger.info(f"Received request: ID={request.request_id}, Text='{request.text_input}', Image={request.image_url}")
        response = await engine.process_request(request)
        return response
    except Exception as e:
        logger.exception(f"Error processing request {request.request_id}")
        raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")

@app.get("/health")
async def health_check():
    return {"status": "healthy"}

import uvicorn
from app.core.config import get_settings

if __name__ == "__main__":
    settings = get_settings()
    uvicorn.run(
        "app.main:app",
        host=settings.server.host,
        port=settings.server.port,
        reload=False, # 生产环境设为False
        log_level=settings.server.log_level
    )

mkdir multimodal-clarification-engine
    cd multimodal-clarification-engine
    # 将上述所有代码文件按项目结构放置到对应位置。

python -m venv venv
    # Linux/Mac
    source venv/bin/activate
    # Windows
    # venv\Scripts\activate

# requirements.txt
    fastapi==0.104.1
    uvicorn[standard]==0.24.0
    pydantic==2.5.0
    pydantic-settings==2.1.0
    transformers==4.36.0
    torch==2.1.0
    Pillow==10.1.0
    requests==2.31.0
    PyYAML==6.0.1

pip install -r requirements.txt

python run.py

    INFO:     Started server process [12345]
    INFO:     Waiting for application startup.
    INFO:     Application startup complete.
    INFO:     Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)

curl -X POST "http://localhost:8000/api/clarify" \
  -H "Content-Type: application/json" \
  -d '{
    "request_id": "test_001",
    "text_input": "帮我推荐一个这个",
    "image_url": "https://images.unsplash.com/photo-1586023492125-27b2c045efd7?w=800",
    "modality": "multimodal"
  }'

curl -X POST "http://localhost:8000/api/clarify" \
  -H "Content-Type: application/json" \
  -d '{
    "request_id": "test_002",
    "text_input": "请为我推荐一款黑色皮质沙发",
    "image_url": null,
    "modality": "text"
  }'

# test_engine.py
import sys
import os
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '.')))

import pytest
from app.schemas.models import UserRequest
from app.core.engine import MultimodalClarificationEngine

@pytest.mark.asyncio
async def test_ambiguous_request_triggers_clarification():
    """测试模糊请求触发澄清"""
    engine = MultimodalClarificationEngine()
    # 注意：这是一个会真实下载图片的测试。可替换为本地测试图片URL或使用Mock。
    request = UserRequest(
        request_id="test_unit_1",
        text_input="这个多少钱？",
        image_url="https://images.unsplash.com/photo-1556228453-efd6c1ff04f6?ixlib=rb-4.0.3&w=200", # 小图加快测试
        modality="multimodal"
    )
    response = await engine.process_request(request)
    # 由于‘这个'模糊且图片可能存在多个对象，置信度很可能低于阈值(0.65)
    # 实际情况取决于模型输出，这里我们断言响应结构正确
    assert response.request_id == "test_unit_1"
    assert isinstance(response.needs_clarification, bool)
    # 如果触发了澄清，clarification字段应有值
    if response.needs_clarification:
        assert response.clarification is not None
        assert len(response.clarification.options) > 0

if __name__ == "__main__":
    import asyncio
    asyncio.run(test_ambiguous_request_triggers_clarification())
    print("Test passed!")