Financiamento aberto da próxima geração: portabilidade de crédito

Queryable Encryption para consenso e privacidade do agente

O armazenamento de registros de consenso em texto simples expõe campos confidenciais a administradores de banco de dados , processos de backup e possíveis violações. Para evitar esses riscos, os regulamentos do Open Finance exigem que as organizações protejam a identidade do consumidor em cada evento do ciclo de vida do consenso : criação, autorização, recuperação de dados e revogação. A criptografia em descanso também protege as configurações do agente de IA — como prompts do sistema e definições de ferramentas — para limitar a exposição da lógica exclusiva.

A Queryable Encryption resolve esse problema criptografando campos sensíveis no nível do driver, garantindo que o servidor nunca veja o texto simples. Os campos que precisam de filtragem podem ser configurados para queries de igualdade. O driver criptografa o valor da query antes de enviá-lo, para que o servidor corresponda ao texto cifrado sem nunca ver o texto simples. Os campos que exigem apenas leitura após descriptografia permanecem criptografados sem suporte a consulta.

A demonstração aplica Queryable Encryption em dois locais:

1. Coleção de permissões (encrypted_consents no backend do Open Finance), com quatro campos criptografados:

   • Consumer.UserName

   • Consumer.UserId

   • Permissions

   • SourceInstitution.InstitutionName

   O campo Consumer.UserName suporta query de igualdade para que os serviços possam listar os autorização de um cliente sem que o banco de dados veja o nome de usuário em texto simples.

2. Coleção de perfil de agente (encrypted_agent_profiles no backend do chatbot), com três campos criptografados:

   • agent_name (pedido de igualdade)

   • system_prompttool_config

Os prompts do agente são carregados do MongoDB criptografado no tempo de execução. A Queryable Encryption gera uma chave de criptografia de dados) separada para cada campo. Ele oferece suporte ao AWS Key Management Service (KMS), Azure Key Vault e Google Cloud KMS como provedores de gerenciamento de chaves. O exemplo a seguir mostra a configuração da conexão criptografada para o backend do OpenFinance:

from pymongo import MongoClient
from pymongo.encryption_options import AutoEncryptionOpts
class EncryptedMongoDBConnection(MongoDBConnection):
    """Subclasses the standard connection — services that type-hint
    MongoDBConnection accept it without modification."""
    def __init__(self, uri: str, auto_encryption_opts: AutoEncryptionOpts):
        self.uri = uri
        self.client = MongoClient(self.uri, auto_encryption_opts=auto_encryption_opts)

As queries de consenso funcionam de forma idêntica ao texto simples – o driver lida com a criptografia e a descriptografia de forma transparente:

# Standard query on a plaintext field — works as usual
consent = consents_collection.find_one({"ConsentId": consent_id})
# Equality query on an encrypted field — same syntax, driver encrypts the filter value
consents = list(consents_collection.find({"Consumer.UserName": user_name}))

A conexão criptografada estende o padrão MongoDBConnection, portanto, cada serviço que digita a classe base a aceita sem modificação.

Classificação inteligente de transações com o MongoDB Atlas pesquisa vetorial

Quando um consumidor compartilha dados bancários externos por meio de um consenso, os registros de transações chegam sem categorias comerciais padronizadas. Strings brutas de fornecedores, como "Restaurante Nobu" ou "UBER TRIP", não contêm metadados de categoria. Sem classificação, a análise de gastos e a pontuação de crédito não podem funcionar.

O Atlas pesquisa vetorial classifica essas transações combinando descrições dos fornecedores com uma coleção de referência de 49 Códigos de categoria de fornecedores (MCC). A demonstração usa o modelo de incorporação voyage-finance-2 da Voyage IA — criado especificamente para textos financeiros — para gerar vetores de dimensão 1024para os dados de referência da MCC e as descrições de transações recebidas.

O exemplo a seguir mostra o pipeline de classificação de pesquisa vetorial:

class MCCClassificationService:
    def __init__(self, connection, db_name, collection_name):
        self.collection = connection.get_collection(db_name, collection_name)
        self.vo = voyageai.Client()
        self.model = "voyage-finance-2"
    def classify_batch(self, transactions):
        # Build query text from merchant name + description
        query_texts = [
            self._build_query_text(txn.get("merchant_name", ""), txn.get("description", ""))
            for txn in transactions
        ]
        # Single batch embedding call — input_type="query" for search queries
        embed_result = self.vo.embed(query_texts, model=self.model, input_type="query")
        # Vector search each embedding against MCC reference codes
        for txn, embedding in zip(transactions, embed_result.embeddings):
            match = self._vector_search(embedding)
            if match:
                txn.update({
                    "MCC": match["MCC"],
                    "CategoryName": match["CategoryName"],
                    "confidence": round(match["score"], 4),
                })
    def _vector_search(self, query_embedding, num_candidates=20, limit=1):
        pipeline = [
            {"$vectorSearch": {
                "index": "mcc_codes_vector_index",
                "path": "embedding",
                "queryVector": query_embedding,
                "numCandidates": num_candidates, "limit": limit,
            }},
            {"$project": {
                "MCC": 1, "MCCDescription": 1, "CategoryId": 1,
                "CategoryName": 1, "score": {"$meta": "vectorSearchScore"}, "_id": 0,
            }},
        ]
        results = list(self.collection.aggregate(pipeline))
        return results[0] if results else None

Os documentos de referência da MCC são incorporados no momento da semente com input_type="document", enquanto as queries de transação usam input_type="query", seguindo a prática recomendada de incorporação assimétrica da Voyage IA para volumes de trabalho de recuperação.

A classificação é efêmera — os resultados são retornados ao agente , mas nunca persistem. Os dados bancários externos são pegados por meio de autorização, e não de propriedade. O agente de portabilidade usa transações classificadas para calcular pontuações de gastos, compará-las com os benchmarks de melhores práticas e gerar ofertas determinísticas de portabilidade de crédito.

Acesso aos dados do Agentic com o servidor do MongoDB MCP

Consultores financeiros e consumidores geralmente precisam de respostas ad-hoc sobre dados da conta: "Qual é o meu saldo total?", "Mostrar minhas últimas 10 transações" ou "Para quais produtos me qualifiquei?". Criar pontos de extremidade de API personalizados para cada query possível não é prático.

O servidor MongoDB MCP expõe coleções MongoDB como ferramentas que os agentes LLM podem invocar diretamente. A demonstração inicia o servidor MCP como um subprocesso na inicialização do aplicativo, conecta-o ao banco de dados leafy_bank no modo somente leitura e passa as ferramentas resultantes para um agente LangGraph por meio de uma sessão persistente.

O exemplo a seguir mostra a integração do servidor MCP:

from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.tools import load_mcp_tools
mcp_client = MultiServerMCPClient({
    "mongodb": {
        "command": "npx",
        "args": ["-y", "mongodb-mcp-server@latest"],
        "transport": "stdio",
        "env": {
            **os.environ,
            "MDB_MCP_CONNECTION_STRING": LEAFY_BANK_MONGODB_URI,
            "MDB_MCP_READ_ONLY": "true",
            "MDB_MCP_DISABLED_TOOLS": disabled_tools,
        },
    }
})
# Persistent session keeps MongoDB connection state across tool calls
async with mcp_client.session("mongodb") as session:
    all_mcp_tools = await load_mcp_tools(session)
    # Pre-connect so the agent never handles connection strings
    connect_tool = next((t for t in all_mcp_tools if t.name == "connect"), None)
    if connect_tool:
        await connect_tool.ainvoke({"connectionString": LEAFY_BANK_MONGODB_URI})
    # Only expose read/query tools to the agent
    allowed_tools = {"find", "aggregate", "count", "list-collections", "collection-schema"}
    mcp_tools = [t for t in all_mcp_tools if t.name in allowed_tools]

O agente do banco rolante recebe essas ferramentas filtradas mais uma ferramenta get_current_user_id que lê o identificador do cliente autenticado a partir da configuração do LangGraph. Ele responde a perguntas de linguagem natural gerando queries MongoDB autoticamente — o agente pode executar as seguintes ações:

• find: use para pesquisas

• aggregate: usar para cálculos

• Collection-schema: use para descoberta.

Nenhum código de ferramenta personalizada é necessário por coleção.

Orquestração de vários agentes com LangGraph

Os fluxos de trabalho de financiamento aberto abrangem domínios distintos: gerenciamento de consenso, análise financeira e consultas internas de dados bancários. Um único agente monocrítico para lidar com os três precisaria de um grande conjunto de ferramentas e um prompt de sistema que abrangesse preocupações conflitantes. A divisão em agentes especializados mantém cada conjunto de ferramentas pequeno e cada prompt focado.

Um agente Supervisor coordena três especialistas:

• Agente de consenso: gerencia fluxos de compartilhamento de dados

• Agente de Portabilidade: Analisa dados externos para ofertas de crédito

• Leafy Banco agente: Consulta dados do Leafy Banco através do servidor MCP.

O LangGraph encaminha cada mensagem do cliente para o especialista apropriado com base na intenção, e o MongoDB Atlas persiste o estado de conversa por meio de coleções de checkpoint .

O exemplo a seguir mostra o roteamento do Supervisor com saída estruturada:

from langgraph.graph import StateGraph, START, END
class RouterDecision(BaseModel):
    next: Literal["consent_agent", "portability_agent", "internal_data_agent", "FINISH"]
    response: str = ""
workflow = StateGraph(AgentState)
workflow.add_node("supervisor", supervisor)
workflow.add_node("consent_agent", consent_agent)
workflow.add_node("portability_agent", portability_agent)
workflow.add_node("internal_data_agent", internal_data_agent)
workflow.add_edge(START, "supervisor")
workflow.add_conditional_edges("supervisor", route_from_supervisor, {
    "consent_agent": "consent_agent",
    "portability_agent": "portability_agent",
    "internal_data_agent": "internal_data_agent",
    "FINISH": END,
})
workflow.add_edge("consent_agent", "supervisor")
workflow.add_edge("portability_agent", "supervisor")
workflow.add_edge("internal_data_agent", "supervisor")
graph = workflow.compile(checkpointer=MongoDBSaver(client=db.client, db_name=DATABASE_NAME))

Fluxos de trabalho regulamentados – aprovações de consenso, revisões de KYC, autorizações de pagamento – exigem checkpoints humanos nos quais um agente deve fazer uma pausa e aguardar uma decisão antes de prosseguir. O mecanismo interrupt() do LangGraph lida com esse requisito serializando o estado completo do grafo para o MongoDB e retornando uma carga útil para o chamador. O fluxo de trabalho é retomado quando o processo externo é concluído:

from langgraph.types import interrupt, Command
# Agent pauses, returns review payload to the calling application
review = interrupt({
    "type": "APPROVAL_REQUIRED",
    "details": approval_details,
})
# Application resumes the workflow after the human decision
await agent.ainvoke(Command(resume=decision), config)

As collections de checkpoint do MongoDB Atlas persistem o estado completo da conversa:

• Histórico de mensagens

• Consensos ativos

• Decisões de roteamento

O fluxo de trabalho sobrevive a interrupções nos últimos segundos (um clique de botão) ou horas (uma revisão de conformidade durante a noite). Cada subagente executa um loop ReAct (motivo → agir → observar) até produzir uma resposta final e, em seguida, retornar o controle ao agente Supervisor para a próxima decisão de roteamento.