La combinación de memoria basada en gráficos con MCP de búsqueda brinda a los agentes de IA un contexto persistente que mejora con el tiempo y al mismo tiempo basa las respuestas en datos web en vivo. Sin memoria, los agentes repiten las mismas búsquedas entre sesiones y pierden las relaciones entre entidades que ya han investigado. Al almacenar los resultados de la búsqueda y las relaciones entre entidades en un gráfico de conocimiento de Neo4j y consultarlo antes de llamar a la búsqueda de Scavio MCP, el agente solo busca información genuinamente nueva mientras construye un gráfico de contexto cada vez más rico.
Requisitos previos
- Python 3.10+
- Neo4j ejecutándose localmente o mediante Aura (el nivel gratuito funciona)
- Controlador neo4j Python instalado (pip install neo4j)
- Clave API de Scavio de scavio.dev
Guia paso a paso
Paso 1: Configurar el almacén de memoria de gráficos
Conéctese a Neo4j y cree el esquema para almacenar entidades, relaciones y resultados de búsqueda con marcas de tiempo.
Python
from neo4j import GraphDatabase
import os
NEO4J_URI = os.environ.get('NEO4J_URI', 'bolt://localhost:7687')
NEO4J_USER = os.environ.get('NEO4J_USER', 'neo4j')
NEO4J_PASS = os.environ.get('NEO4J_PASS', 'password')
driver = GraphDatabase.driver(NEO4J_URI, auth=(NEO4J_USER, NEO4J_PASS))
def init_schema():
with driver.session() as s:
s.run('CREATE CONSTRAINT IF NOT EXISTS FOR (e:Entity) REQUIRE e.name IS UNIQUE')
s.run('CREATE CONSTRAINT IF NOT EXISTS FOR (s:SearchResult) REQUIRE s.url IS UNIQUE')
s.run('CREATE INDEX IF NOT EXISTS FOR (s:SearchResult) ON (s.query)')
print('Graph schema initialized')
init_schema()Paso 2: Cree la función de búsqueda con reconocimiento de memoria
Cree una función de búsqueda que verifique primero la memoria del gráfico y solo llame a la API de Scavio para consultas que no se hayan visto recientemente.
Python
import requests
from datetime import datetime, timedelta
H = {'x-api-key': os.environ['SCAVIO_API_KEY'], 'Content-Type': 'application/json'}
CACHE_TTL_HOURS = 24
def search_with_memory(query):
# Check graph memory first
with driver.session() as s:
cached = s.run(
'MATCH (sr:SearchResult {query: $q}) '
'WHERE sr.fetched_at > datetime() - duration({hours: $ttl}) '
'RETURN sr.title AS title, sr.url AS url, sr.snippet AS snippet '
'ORDER BY sr.position LIMIT 10',
q=query, ttl=CACHE_TTL_HOURS
).data()
if cached:
print(f'Memory hit: {len(cached)} results for "{query}"')
return cached
# Cache miss: search live
data = requests.post('https://api.scavio.dev/api/v1/search',
headers=H, json={'query': query, 'country_code': 'us'}).json()
results = data.get('organic_results', [])
# Store in graph
with driver.session() as s:
for r in results:
s.run(
'MERGE (sr:SearchResult {url: $url}) '
'SET sr.title = $title, sr.snippet = $snippet, '
'sr.query = $query, sr.position = $pos, sr.fetched_at = datetime()',
url=r.get('link', ''), title=r.get('title', ''),
snippet=r.get('snippet', ''), query=query, pos=r.get('position', 0)
)
print(f'Live search: {len(results)} results stored for "{query}"')
return resultsPaso 3: Agregar extracción de entidades y seguimiento de relaciones
Extraiga entidades de los resultados de búsqueda y cree relaciones en el gráfico para que el agente acumule conocimientos entre sesiones.
Python
import re
def extract_entities(text, entity_types=None):
# Simple pattern-based extraction (replace with NER model for production)
entities = set()
# Capitalized multi-word phrases (likely proper nouns)
for match in re.findall(r'\b([A-Z][a-z]+(?: [A-Z][a-z]+)+)\b', text):
entities.add(match)
return list(entities)
def store_entities_from_results(query, results):
with driver.session() as s:
topic = s.run(
'MERGE (e:Entity {name: $name}) '
'SET e.type = "topic", e.last_searched = datetime() '
'RETURN e', name=query
).single()
for r in results:
text = f"{r.get('title', '')} {r.get('snippet', '')}"
entities = extract_entities(text)
for ent in entities:
s.run(
'MERGE (e:Entity {name: $name}) '
'WITH e '
'MATCH (t:Entity {name: $topic}) '
'MERGE (t)-[:RELATED_TO]->(e)',
name=ent, topic=query
)
print(f'Stored entities for "{query}"')
# Usage
results = search_with_memory('LangGraph agent tutorial')
store_entities_from_results('LangGraph agent tutorial', results)Paso 4: Consulta el gráfico de conocimiento para conocer el contexto
Antes de buscar, pregunte al gráfico qué sabe ya el agente sobre un tema para crear un contexto más rico para el LLM.
Python
def get_context_from_memory(topic, depth=2):
with driver.session() as s:
# Get related entities up to N hops away
related = s.run(
'MATCH (e:Entity {name: $name})-[:RELATED_TO*1..' + str(depth) + ']-(r:Entity) '
'RETURN DISTINCT r.name AS name, r.type AS type '
'LIMIT 20',
name=topic
).data()
# Get recent search results for related topics
context_results = s.run(
'MATCH (e:Entity {name: $name})-[:RELATED_TO*1..2]-(r:Entity) '
'MATCH (sr:SearchResult) WHERE sr.query CONTAINS r.name '
'RETURN sr.title AS title, sr.snippet AS snippet, sr.url AS url '
'ORDER BY sr.fetched_at DESC LIMIT 10',
name=topic
).data()
context = {
'known_entities': [r['name'] for r in related],
'related_results': context_results
}
print(f'Context: {len(related)} entities, {len(context_results)} cached results')
return context
ctx = get_context_from_memory('LangGraph agent tutorial')
print(f'Known entities: {ctx["known_entities"][:5]}')Ejemplo en Python
Python
import os, requests, re
from neo4j import GraphDatabase
H = {'x-api-key': os.environ['SCAVIO_API_KEY'], 'Content-Type': 'application/json'}
driver = GraphDatabase.driver(
os.environ.get('NEO4J_URI', 'bolt://localhost:7687'),
auth=(os.environ.get('NEO4J_USER', 'neo4j'), os.environ.get('NEO4J_PASS', 'password')))
def init_schema():
with driver.session() as s:
s.run('CREATE CONSTRAINT IF NOT EXISTS FOR (e:Entity) REQUIRE e.name IS UNIQUE')
s.run('CREATE CONSTRAINT IF NOT EXISTS FOR (sr:SearchResult) REQUIRE sr.url IS UNIQUE')
def search_with_memory(query, ttl_hours=24):
with driver.session() as s:
cached = s.run(
'MATCH (sr:SearchResult {query: $q}) '
'WHERE sr.fetched_at > datetime() - duration({hours: $ttl}) '
'RETURN sr.title AS title, sr.url AS url, sr.snippet AS snippet LIMIT 10',
q=query, ttl=ttl_hours).data()
if cached:
return {'source': 'memory', 'results': cached}
data = requests.post('https://api.scavio.dev/api/v1/search',
headers=H, json={'query': query, 'country_code': 'us'}).json()
results = data.get('organic_results', [])
with driver.session() as s:
for r in results:
s.run('MERGE (sr:SearchResult {url: $url}) '
'SET sr.title=$t, sr.snippet=$sn, sr.query=$q, sr.fetched_at=datetime()',
url=r.get('link',''), t=r.get('title',''), sn=r.get('snippet',''), q=query)
return {'source': 'live', 'results': results}
def get_graph_context(topic, depth=2):
with driver.session() as s:
return s.run(
'MATCH (e:Entity {name:$n})-[:RELATED_TO*1..'+str(depth)+']-(r) '
'RETURN DISTINCT r.name AS name LIMIT 20', n=topic).data()
init_schema()
result = search_with_memory('LangGraph agent patterns 2026')
print(f"Source: {result['source']}, Results: {len(result['results'])}")Ejemplo en JavaScript
JavaScript
const neo4j = require('neo4j-driver');
const driver = neo4j.driver(
process.env.NEO4J_URI || 'bolt://localhost:7687',
neo4j.auth.basic(process.env.NEO4J_USER || 'neo4j', process.env.NEO4J_PASS || 'password'));
const H = {'x-api-key': process.env.SCAVIO_API_KEY, 'Content-Type': 'application/json'};
async function searchWithMemory(query, ttlHours = 24) {
const session = driver.session();
try {
const cached = await session.run(
'MATCH (sr:SearchResult {query: $q}) '
+ 'WHERE sr.fetched_at > datetime() - duration({hours: $ttl}) '
+ 'RETURN sr.title AS title, sr.url AS url LIMIT 10',
{q: query, ttl: neo4j.int(ttlHours)});
if (cached.records.length > 0) {
return {source: 'memory', results: cached.records.map(r => r.toObject())};
}
const data = await fetch('https://api.scavio.dev/api/v1/search', {
method: 'POST', headers: H,
body: JSON.stringify({query, country_code: 'us'})
}).then(r => r.json());
const results = data.organic_results || [];
for (const r of results) {
await session.run(
'MERGE (sr:SearchResult {url: $url}) SET sr.title=$t, sr.query=$q, sr.fetched_at=datetime()',
{url: r.link || '', t: r.title || '', q: query});
}
return {source: 'live', results};
} finally { await session.close(); }
}
searchWithMemory('LangGraph agent patterns').then(r =>
console.log(\`Source: \${r.source}, Results: \${r.results.length}\`));Salida esperada
JSON
Graph schema initialized
Live search: 10 results stored for "LangGraph agent patterns 2026"
Source: live, Results: 10
# Second run:
Memory hit: 10 results for "LangGraph agent patterns 2026"
Source: memory, Results: 10