Análise de rendimento Agentic com MongoDB

Incorporações multimodais para Defeitos de Bolacha

Esta solução usa o modelo route-multimodal-3 da Voyage AI para gerar incorporações que combinam imagens de erro de disco com contexto textual. Isso permite a pesquisa de similaridade semântica em padrões visuais e texto descritivo.

Incorporando serviço

A classe EmbeddingService lida com a geração de incorporação usando o cliente Voyage AI :

import voyageai
from PIL import Image
import base64
import io
class EmbeddingService:
    def __init__(self):
        self.voyage_client = voyageai.Client(api_key=os.getenv("VOYAGE_API_KEY"))
        self.multimodal_model = "voyage-multimodal-3"
        self.embedding_dimension = 1024
    async def generate_image_embedding(
        self,
        image_data: str,
        text_context: str = None
    ) -> List[float]:
        """Generate multimodal embedding from image and text."""
        # Decode base64 image to PIL Image
        image_bytes = base64.b64decode(image_data)
        pil_image = Image.open(io.BytesIO(image_bytes))
        # Combine text and image inputs
        inputs = []
        if text_context:
            inputs.append(text_context)
        inputs.append(pil_image)
        # Generate embedding
        result = self.voyage_client.multimodal_embed(
            inputs=[inputs],
            model=self.multimodal_model,
            input_type="document"
        )
        return result.embeddings[0]

Processando Defeitos de Bolacha

Para cada documento de erro de pasta, o pipeline executa as seguintes ações:

1. Constrói conteúdo de texto a partir de características observáveis (ID do timestamp, padrão de erro, material, rendimento, descrição do consumidor).

2. Busca a imagem do mapa de tinta do Amazon S3.

3. Gera uma incorporação multimodal combinando ambas as entradas.

4. Armazena o vetor dimensional 1024no campo embedding.

# Build text from observable facts only (not suspected causes)
text_content = f"Wafer ID: {wafer['wafer_id']} "
text_content += f"Defect pattern: {wafer['defect_summary']['defect_pattern']} "
text_content += f"Equipment: {wafer['process_context']['equipment_used'][0]} "
text_content += f"Yield: {wafer['defect_summary']['yield_percentage']}%"
# Get image data
image_data = wafer["ink_map"]["thumbnail_base64"]
# Generate multimodal embedding
embedding = await embedding_service.generate_image_embedding(
    image_data=image_data,
    text_context=text_content
)
# Store in document
await db.wafer_defects.update_one(
    {"_id": wafer["_id"]},
    {"$set": {
        "embedding": embedding,
        "embedding_type": "multimodal",
        "embedding_model": "voyage-multimodal-3"
    }}
)

Índice de Vector Search

Crie um índice de pesquisa vetorial na collection wafer_defects no MongoDB Atlas:

{
  "name": "wafer_defects_vector_search",
  "type": "vectorSearch",
  "definition": {
    "fields": [
      {
        "path": "embedding",
        "type": "vector",
        "numDimensions": 1024,
        "similarity": "cosine"
      }
    ]
  }
}

Isso permite que o agente de causa raiz encontre falhas históricas semelhantes usando a pesquisa vetorial, mesmo quando a nova pasta não tem causa raiz conhecida.

Ferramentas do Agente

Ferramentas são funções específicas do domínio que permitem ao agente interagir com o MongoDB Atlas. Eles consultam dados de sensores, realizam pesquisa semântica e recuperam padrões históricos. Cada ferramenta retorna dados estruturados que o LLM analisa para gerar relatórios IRA.

O código a seguir mostra como registrar uma ferramenta para o Agente de Causa Raiz usando o decoração @tool do LangChain. Neste exemplo, a ferramenta usa a pesquisa vetorial para encontrar padrões semelhantes de falhas de bocha.

from langchain_core.tools import tool
@tool
async def query_wafer_info(
    wafer_id: str,
    include_similar_patterns: bool = True,
    similarity_limit: int = 3
) -> Dict[str, Any]:
    """
    Get wafer defect details and find similar historical patterns.
    Returns the wafer data plus similar past defects with known root causes.
    """
    db = _get_db()
    wafer = await db.wafer_defects.find_one({"wafer_id": wafer_id})
    if not wafer:
        return {"error": f"Wafer {wafer_id} not found"}
    similar_patterns = None
    if include_similar_patterns and "embedding" in wafer:
        pipeline = [
            {
                "$vectorSearch": {
                    "index": "wafer_defects_vector_index",
                    "path": "embedding",
                    "queryVector": wafer["embedding"],
                    "numCandidates": 100,
                    "limit": similarity_limit + 1
                }
            },
            {"$match": {"wafer_id": {"$ne": wafer_id}}},
            {"$addFields": {"similarity_score": {"$meta": "vectorSearchScore"}}},
            {"$limit": similarity_limit}
        ]
        results = await db.wafer_defects.aggregate(pipeline).to_list(length=None)
        similar_patterns = [
            {
                "wafer_id": r.get("wafer_id"),
                "description": r.get("description"),
                "root_cause": r.get("root_cause"),
                "similarity_score": round(r.get("similarity_score", 0), 4)
            }
            for r in results
        ]
    return {
        "wafer": wafer,
        "similar_historical_patterns": similar_patterns
    }
# Tool registry
TOOLS = [query_alerts, query_wafer_info, query_time_series_data, vector_search_knowledge_base]

O agente de causa raiz usa as seguintes ferramentas para investigar Excursões:

1. query_alerts

   • Recupera alertas recentes filtrados por hardware, gravidade ou janela de tempo.

   • Retorna detalhes de violação, bocha afetadas e dados do sensor de origem.

2. query_wafer_info

   • Obtém detalhes de erros de ondulação, incluindo porcentagem de rendimento, padrão de erro e gravidade.

   • Executa pesquisa vetorial multimodal para encontrar falhas históricas semelhantes com causa raiz conhecida.

3. query_time_series_data

   • Consulta a telemetria do sensor em torno de uma janela de tempo específica.

   • Retorna estatísticas agregadas (min, max, avg) para reduzir o uso de token.

   • Identifica anomalias de sensor correlacionadas com eventos de erro.

4. vector_search_knowledge_base

   • Pesquisa relatórios históricos de IRA e documentação técnica usando incorporações semânticas.

   • Retorna documentos correspondentes com títulos, causas raiz e ações corretivas.

   • Ajuda o agente a consultar soluções anteriores para falhas semelhantes.

Você pode expandir este conjunto de ferramentas para corresponder aos processos da sua fab. Por exemplo, adicione ferramentas para consultar registros de manutenção de equipes, verificar parâmetros de receita ou recuperar anotações de deslocamento do operador.

Memória de agente

Para que os agentes funcionem de forma eficaz, eles precisam de memória para armazenar as etapas de contexto e argumentos. Esse recurso permite que os agentes:

• Mantenha a sequência dentro de uma pesquisa.

• Lembre-se das etapas anteriores e saídas da ferramenta.

• Crie contexto entre as interações do usuário.

Nesta arquitetura, o MongoDB Atlas armazena toda a memória do agente . A memória consiste nos seguintes tipos:

Memória de curto prazo: armazena o estado intermediário à medida que o agente se move pela pesquisa. Essa memória garante que, se um processo for interrompido, ele possa ser retomado sem perder o progresso. As seguintes coleções armazenam esse tipo de memória:

• checkpoints: captura o estado do agente em cada etapa de argumentos.

• checkpoint_writes: registra as chamadas de ferramenta e seus resultados.

Memória de longo prazo: armazena dados históricos que informam as pesquisas atuais. Os agentes recuperam esses dados usando pesquisa vetorial, garantindo que o contexto histórico impulsione o argumento. As coleções incluem:

• wafer_defects: Dados de inspeção de Wáfer com incorporações multimodais para pesquisa de similaridade.

• historical_knowledge: relatórios IRA, documentação técnica e conhecimento Tribal.

• alerts: alertas ativos e resolvidos com detalhes de violação e dados de origem.

• process_sensor_ts: Telemetria de sensor de série temporal para análise de correlação.

Para configurar a memória de curto prazo, use a classe MongoDBSaver do LangGraph. Esta classe escreve o progresso do agente nas collections checkpoints da seguinte forma:

from langgraph.checkpoint.mongodb import MongoDBSaver
from pymongo import MongoClient
mongo_client = MongoClient(os.getenv("MONGODB_URI"))
checkpointer = MongoDBSaver(mongo_client, "smf-yield-defect")

Esta configuração ativa os recursos de memória e tolerância a falhas para o agente raiz.

Gráfico de estado do agente

Um gráfico de estado modela fluxos de trabalho como nós e bordas. Cada nó representa uma etapa de reflexão, chamada de ferramenta ou checkpoint. As bordas definem transições entre essas etapas. Os gráficos de estado tornam os fluxos de trabalho explícitos, repetíveis e resilientes.

Nesta solução, o LangGraph permite que o gráfico do estado coordene o Agente de Causa Raiz e suas ferramentas. O agente segue um padrão ReAct (Reasoning + Aging):

1. Motivo: o LLM analisa o estado atual e decide a próxima ação.

2. Agir: o agente chama uma ferramenta para recuperar dados do MongoDB.

3. Observar: O agente processa a saída da ferramenta e atualiza seu argumento.

4. Repetir: o ciclo continua até que o agente tenha provas suficientes.

Essa arquitetura garante os seguintes recursos:

• O agente pode se ramificar com base em descobertas, como padrões semelhantes encontrados versus nenhuma correspondência.

• Cada etapa escreve na memória e lê dela automaticamente.

• Os engenheiros podem retomar conversas ou auditar a cadeia de lógica.

O código a seguir cria um agente ReAct com o checkpoint do MongoDB :

from langgraph.prebuilt import create_react_agent
from langchain_aws import ChatBedrock
async def create_rca_agent():
    """Create LangGraph agent with MongoDB checkpointing."""
    # Initialize LLM
    llm = ChatBedrock(
        model_id="anthropic.claude-3-5-sonnet-20241022-v2:0",
        region_name=os.getenv("AWS_REGION", "us-east-1")
    )
    # Initialize MongoDB checkpointer
    mongo_client = MongoClient(os.getenv("MONGODB_URI"))
    checkpointer = MongoDBSaver(mongo_client, "smf-yield-defect")
    # System prompt
    system_prompt = """You are an expert semiconductor yield engineer.
    When investigating alerts:
    1. First, query the alert details
    2. Get wafer defect information and similar historical patterns
    3. Query sensor data around the alert time
    4. Search the knowledge base for similar RCA reports
    5. Synthesize findings into a structured root cause analysis
    Always cite evidence from the tools."""
    # Create agent
    agent = create_react_agent(
        model=llm,
        tools=TOOLS,
        checkpointer=checkpointer,
        prompt=system_prompt
    )
    return agent

Com essa configuração, você pode rastrear, retomar e depurar todo o fluxo de trabalho da pesquisa.

Fluxo de trabalho de ponta a ponta

O seguinte descreve como o sistema processa uma viagem:

Você pode expandir e personalizar esse fluxo de trabalho com os seguintes recursos:

• Correção automatizada: acione o isolamento do hardware ou os ajustes da receita com base nas descobertas da RB.

• Alertas preditivos: use padrões históricos para avisar antes que os limites sejam violados.

• Correlação de várias ferramentas: adicione ferramentas para consultar parâmetros de receitas, registros de salas ou cronogramas de manutenção.

Como as ferramentas, a memória e a orquestração de gráficos são modulares, você pode adicionar novos recursos sem interromper os fluxos de trabalho existentes.

Coleções principais

Esta solução utiliza as seguintes collections para armazenar dados:

sensor_events: Eventos de sensor em tempo real para monitoramento do Change Stream. Essa coleção regular permite que o Sistema de Detecção de Excursões observe violações de limites em tempo real.

alerts: Excursões ativas e violações de limites que acionam o Agente de Causa Raiz. Cada alerta captura os detalhes da violação, a bocha afetada e os dados do sensor de origem. O status transita de "aberto" para "reconhecido" para "resolvido".

wafer_defects: Dados de inspeção de Wáfer com incorporações multimodais para pesquisa semântica. Cada documento inclui padrões de erros, porcentagens de rendimento, níveis de severidade e a incorporação combinada de imagem e texto gerada pela IA do Voyage.

historical_knowledge: Relatórios de rca e documentação técnica armazenados com incorporações vetoriais. Os agentes pesquisam essa coleção para encontrar incidentes anteriores semelhantes, procedimentos de solução de problemas e ações corretivas testadas.

process_context: Metadados do processo de fabrica

checkpoints: Estado do agente capturado em cada etapa de argumentos pelo MongoDBSaver da LangGraph para permitir a persistência da conversa, a retomada da sessão e as faixas de auditar .

process_sensor_ts: Telemetria do sensor de processo armazenada como uma coleção de séries temporais para análise histórica eficiente. As coleções de séries temporais armazenam e consultam com eficiência milhões de leituras. Elas preservam metadados contextuais, como ID do dispositivo, ID do lote e etapa do processo.

O exemplo a seguir mostra um documento de amostra na coleção process_sensor_ts:

{
  "timestamp": {
    "$date": "2025-01-24T10:30:00.000Z"
  },
  "equipment_id": "CMP_TOOL_01",
  "metrics": {
    "particle_count": 1234,
    "temperature": 68.5,
    "rf_power": 1502.3,
    "chamber_pressure": 5.2
  },
  "metadata": {
    "lot_id": "LOT_2025_001",
    "wafer_id": "W_004_16",
    "process_step": "Oxide CMP"
  }
}

O documento de série temporal inclui os seguintes campos:

• timestamp: O carimbo de data/hora da leitura

• equipment_id: O identificador da ferramenta de origem

• metrics: Valores de sensor numérico para contagem de amostras, temperatura, potência de RF e pressão da sala

• metadata: Tags contextuais para lote, pasta e etapa do processo

O exemplo a seguir mostra um documento de amostra na coleção wafer_defects:

{
  "_id": "W_CMP_001",
  "wafer_id": "W_CMP_001",
  "lot_id": "LOT_2025_001",
  "inspection_timestamp": { "$date": "2025-01-24T10:30:00Z" },
  "description": "Edge-concentrated particle contamination from slurry degradation",
  "defect_summary": {
    "defect_pattern": "edge_cluster",
    "severity": "critical",
    "yield_percentage": 72.5,
    "failed_dies": 22,
    "total_dies": 80
  },
  "process_context": {
    "equipment_used": ["CMP_TOOL_01"],
    "last_process_step": "Oxide CMP",
    "recipe_id": "CMP_STD_01",
    "slurry_batch": "SLR-2025-0142"
  },
  "ink_map": {
    "thumbnail_base64": "iVBORw0KGgo...",
    "thumbnail_size": { "width": 200, "height": 200 },
    "full_image_url": "s3://bucket/wafers/W_CMP_001.png"
  },
  "embedding": [0.123, -0.456, ...]
}

O documento de erro do Whafer inclui os seguintes campos:

• wafer_id e lot_id: vincula o erro ao contexto de produção

• description: contém a análise de causa raiz para bocha históricas

• defect_summary: captura o tipo de padrão, a gravidade, o impacto no rendimento e as contagens de matrizes

• process_context: Rastreia máquinas, receitas e materiais para análise de correlação

• ink_map: Armazena a visualização do mapa de javali (thumbdown para exibição, URL S3 para imagem completa)

• embedding: contém o vetor multimodal 1024-dimensional para pesquisa de similaridade