Fundamentos de la integración multimodal

Los sistemas tradicionales de preguntas y respuestas basados en texto presentan limitaciones cuando los usuarios envía imágenes con consultas. En plataformas educativas, estudiantes pueden subir fotografías de problemas geométricos; en comercio electrónico, clientes adjuntan fotos de productos con preguntas específicas. Un sistema que combina comprensión visual con capacidades conversacionales resuelve estos escenarios.

OFA (One-For-All) es un modelo preentrenado unificado que aborda múltiples tareas mediante un marco de secuencia a secuencia: descripción de imágenes, razonamiento visual y respuesta a preguntas visuales. LangChain proporciona la infraestructura para orquestar componentes, permitting integrar OFA como herramienta dentro de flujos conversacionales más amplios.

Comparación de enfoques

Configuración del entorno

Requisitos mínimos: Python 3.8-3.9, 8GB RAM, 10GB espacio en disco. GPU NVIDIA opcional pero recomendado.

# Crear entorno virtual
python3 -m venv venv_multimodal
source venv_multimodal/bin/activate

# Instalar dependencias
pip install torch torchvision --index-url https://download.pytorch.org/whl/cu118
pip install modelscope langchain openai pillow python-dotenv faiss-cpu

Carga de modelos OFA

ModelScope facilita la descarga y carga de los modelos preentrenados de OFA:

from modelscope.pipelines import pipeline
from modelscope.utils.constant import Tasks

# Inicializar pipeline de captioning
caption_generator = pipeline(
    task=Tasks.image_captioning,
    model='damo/ofa_image-caption_coco_large_en'
)

# Inicializar pipeline de entailment visual
visual_reasoner = pipeline(
    task=Tasks.visual_entailment,
    model='damo/ofa_visual-entailment_snli-ve_large_en'
)

Configuración del motor conversacional

import os
from dotenv import load_dotenv
from langchain.llms import OpenAI

load_dotenv()

conversational_engine = OpenAI(
    temperature=0.6,
    max_tokens=450,
    openai_api_key=os.environ.get("OPENAI_API_KEY")
)

Implementación del análisis visual como herramienta LangChain

Encapsulamos la funcionalidad de OFA en una herramienta compatible con el sistema de agentes de LangChain:

from langchain.tools import BaseTool
from PIL import Image
import requests
from io import BytesIO

class VisualContentInterpreter(BaseTool):
    """Herramienta para interpretar contenido visual mediante OFA"""
    
    name: str = "visual_interpreter"
    description: str = "Interpreta imágenes generando descripciones y evaluando relaciones lógicas con texto"
    
    def _load_image(self, source: str) -> Image.Image:
        if source.startswith("http"):
            resp = requests.get(source)
            return Image.open(BytesIO(resp.content))
        return Image.open(source)
    
    def _run(self, img_source: str, query_text: str = "") -> dict:
        img = self._load_image(img_source)
        output = {}
        
        # Generar descripción
        cap_output = caption_generator(img)
        output["description"] = cap_output["caption"][0]
        
        # Evaluar relación lógica si hay consulta
        if query_text:
            ve_input = {"image": img, "text": query_text}
            ve_output = visual_reasoner(ve_input)
            output["logic_relation"] = ve_output["label"]
            output["confidence_score"] = ve_output["score"]
        
        return output
    
    async def _arun(self, img_source: str, query_text: str = ""):
        return self._run(img_source, query_text)

visual_tool = VisualContentInterpreter()

Construcción del agente conversacional

from langchain.agents import initialize_agent, AgentType
from langchain.prompts import PromptTemplate
from langchain.memory import ConversationBufferMemory

dialog_memory = ConversationBufferMemory(
    memory_key="dialogue_log", 
    return_messages=True
)

agent_instructions = """Eres un asistente que comprende imágenes y responde preguntas.

Cuando el usuario mencione imágenes, usa visual_interpreter para analizarlas.
La herramienta devuelve descripciones y relaciones lógicas con el texto.

Historial de conversación:
{dialogue_log}

Consulta del usuario: {input}

Determina si necesitas analizar alguna imagen y responde adecuadamente."""

agent_prompt = PromptTemplate(
    input_variables=["dialogue_log", "input"],
    template=agent_instructions
)

qa_agent = initialize_agent(
    tools=[visual_tool],
    llm=conversational_engine,
    agent=AgentType.CONVERSATIONAL_REACT_DESCRIPTION,
    memory=dialog_memory,
    verbose=True
)

Función unificada de procesamiento de consultas

def handle_user_query(question: str, img_ref: str = None) -> str:
    """Procesa consultas de usuario con o sin imágenes"""
    
    if img_ref:
        print(f"Analizando imagen: {img_ref}")
        analysis = visual_tool.run(img_ref, question)
        
        context_prompt = f"""
Resultados del análisis visual:
- Descripción: {analysis.get('description', 'N/A')}
- Relación lógica: {analysis.get('logic_relation', 'N/A')} 
- Confianza: {analysis.get('confidence_score', 'N/A')}

Pregunta: {question}

Responde basándote en el análisis visual."""
        
        return qa_agent.run(context_prompt)
    
    if any(kw in question.lower() for kw in ["imagen", "foto", "image"]):
        return "Detecté que mencionas una imagen. Proporciona la ruta o URL para analizarla."
    
    return qa_agent.run(question)

Casos de uso prácticos

Resolución de problemas geométricos

geometry_answer = handle_user_query(
    question="¿Cuál es el valor del ángulo desconocido?",
    img_ref="https://ejemplo.com/triangulo.jpg"
)
print(geometry_answer)

Consultas de productos en e-commerce

product_answer = handle_user_query(
    question="¿Qué material es este objeto y está disponible?",
    img_ref="producto_001.jpg"
)

Verificación de consistencia imagen-texto

def verify_text_image_consistency(img_path: str, claim: str) -> str:
    result = visual_tool.run(img_path, claim)
    relation = result.get("logic_relation", "neutral")
    
    status_map = {
        "entailment": "✓ La imagen coincide con la descripción",
        "contradiction": "✗ La imagen contradice la descripción",
        "neutral": "○ Relación indeterminada"
    }
    return status_map.get(relation, "○ Relación indeterminada")

consistency = verify_text_image_consistency(
    "subida_usuario.jpg", 
    "A beautiful sunset at the beach"
)

Optimizaciones avanzadas

Análisis multi-región de imágenes

def multi_region_analysis(img_path: str, query: str = "") -> dict:
    """Analiza múltiples regiones de una imagen grande"""
    from PIL import Image
    
    img = Image.open(img_path)
    w, h = img.size
    analyses = []
    
    # Análisis completo
    analyses.append(visual_tool._run(img_path, query))
    
    # Análisis de regiones si la imagen es grande
    if w > 800 and h > 800:
        regions = [
            img.crop((0, 0, w//2, h//2)),
            img.crop((w//2, 0, w, h//2)),
            img.crop((0, h//2, w//2, h)),
            img.crop((w//2, h//2, w, h))
        ]
        for idx, region in enumerate(regions):
            temp_path = f"region_{idx}.jpg"
            region.save(temp_path)
            analyses.append(visual_tool._run(temp_path, query))
    
    combined_desc = " | ".join(
        a["description"] for a in analyses if "description" in a
    )
    
    return {"combined_description": combined_desc, "detail_list": analyses}

Gestión de contexto con memoria de imágenes

import time

class VisualContextMemory:
    """Memoria que rastrea la imagen actualmente en discusión"""
    
    def __init__(self):
        self._active_image = None
        self._image_log = []
        self._dialog = ConversationBufferMemory()
    
    def register_image(self, path: str, desc: str):
        self._active_image = {
            "path": path,
            "desc": desc,
            "timestamp": time.time()
        }
        self._image_log.append(self._active_image)
    
    def build_contextualized_prompt(self, user_query: str) -> str:
        if not self._active_image:
            return user_query
        
        return f"""
Imagen en contexto: {self._active_image['desc']}
Registrada: {time.ctime(self._active_image['timestamp'])}

Consulta: {user_query}

Responde considerando la imagen actual."""

Integración con base de conocimiento vectorial

from langchain.vectorstores import FAISS
from langchain.embeddings import OpenAIEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter

class KnowledgeAugmentedQA:
    """Sistema QA con recuperación de conocimiento externo"""
    
    def __init__(self, docs_path: str):
        with open(docs_path, 'r', encoding='utf-8') as f:
            raw_text = f.read()
        
        splitter = RecursiveCharacterTextSplitter(
            chunk_size=800, chunk_overlap=80
        )
        chunks = splitter.split_text(raw_text)
        
        self.vector_db = FAISS.from_texts(
            chunks, OpenAIEmbeddings()
        )
    
    def retrieve_context(self, query: str, top_k: int = 3) -> str:
        matches = self.vector_db.similarity_search(query, k=top_k)
        return "\n---\n".join(d.page_content for d in matches)
    
    def answer_with_context(self, question: str, img_data: dict = None):
        context = self.retrieve_context(question)
        
        img_section = ""
        if img_data:
            img_section = f"Información visual: {img_data.get('description', '')}"
        
        full_prompt = f"""Contexto recuperado:
{context}

{img_section}

Pregunta: {question}

Proporciona una respuesta fundamentada."""
        
        return qa_agent.run(full_prompt)

Caché de resultados de análisis

import hashlib
from functools import lru_cache

@lru_cache(maxsize=128)
def cached_visual_analysis(img_path: str, query: str = "") -> dict:
    """Análisis visual con caché para evitar recálculos"""
    with open(img_path, 'rb') as f:
        file_hash = hashlib.md5(f.read()).hexdigest()
    
    # En producción, usar Redis u otro almacenamiento persistente
    return visual_tool.run(img_path, query)

Interfaz interactiva de demostración

def run_interactive_session():
    """Ejecuta una sesión interactiva en consola"""
    print("=" * 55)
    print("Sistema Multimodal de Preguntas y Respuestas")
    print("=" * 55)
    print("\nComandos disponibles:")
    print("- Pregunta de texto directo")
    print("- 'img:/ruta | pregunta' para análisis con imagen")
    print("- 'salir' para terminar")
    
    while True:
        user_line = input("\n>>> ").strip()
        
        if user_line.lower() in ['salir', 'exit', 'quit']:
            print("Sesión finalizada.")
            break
        
        if user_line.startswith("img:"):
            parts = user_line.split("|", 1)
            if len(parts) == 2:
                img_ref = parts[0].replace("img:", "").strip()
                question = parts[1].strip()
                response = handle_user_query(question, img_ref)
            else:
                print("Formato: img:/ruta/a/imagen.jpg | tu pregunta")
                continue
        else:
            response = handle_user_query(user_line)
        
        print(f"\nRespuesta: {response}")

if __name__ == "__main__":
    run_interactive_session()