Los agentes LangGraph obtienen un valor significativo cuando combinan la memoria persistente con la base de búsqueda en vivo. La memoria permite al agente recordar interacciones pasadas y preferencias del usuario, mientras que la búsqueda evita las alucinaciones al inyectar datos en tiempo real. Este tutorial crea un agente LangGraph que mantiene el estado de la conversación en todos los turnos, utiliza la API de búsqueda de Scavio como herramienta y decide cuándo buscar y cuándo confiar en la memoria. La herramienta de búsqueda cuesta 0,005 dólares por llamada, por lo que incluso los agentes conversadores se quedan con unos pocos dólares al mes.
Requisitos previos
- Python 3.10+ instalado
- Paquetes langgraph y langchain-openai instalados
- Clave API de OpenAI para el LLM
- Clave API de Scavio para la base de búsqueda
Guia paso a paso
Paso 1: Definir el esquema de estado del agente
LangGraph utiliza el estado escrito para gestionar el flujo de la conversación. Defina un estado que contenga mensajes, resultados de búsqueda y un almacén de memoria.
from typing import TypedDict, Annotated, Sequence
from langchain_core.messages import BaseMessage
import operator
class AgentState(TypedDict):
messages: Annotated[Sequence[BaseMessage], operator.add]
search_results: str
memory_context: strPaso 2: Construya el nodo de la herramienta de búsqueda
Cree un nodo que llame a la API de Scavio cuando el agente decida que necesita información actual. El nodo recibe el estado y devuelve resultados de búsqueda actualizados.
import requests, os
API_KEY = os.environ['SCAVIO_API_KEY']
def search_node(state: AgentState) -> dict:
last_msg = state['messages'][-1].content
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': API_KEY, 'Content-Type': 'application/json'},
json={'query': last_msg, 'country_code': 'us'})
resp.raise_for_status()
results = resp.json().get('organic_results', [])[:5]
context = '\n'.join(f'- {r["title"]}: {r.get("snippet", "")}' for r in results)
return {'search_results': context}Paso 3: Construya el nodo de gestión de memoria
Cree un almacén de memoria simple que conserve los hechos clave de la conversación. El agente puede consultarlos en turnos futuros sin tener que volver a buscarlos.
memory_store = {}
def memory_node(state: AgentState) -> dict:
# Extract facts from the latest exchange
messages = state['messages']
if len(messages) >= 2:
user_msg = messages[-2].content if len(messages) >= 2 else ''
ai_msg = messages[-1].content if messages else ''
# Store key topic as memory
key = user_msg[:50]
memory_store[key] = ai_msg[:200]
# Build memory context from recent entries
recent = list(memory_store.items())[-5:]
ctx = '\n'.join(f'Previously discussed: {k} -> {v[:80]}' for k, v in recent)
return {'memory_context': ctx}Paso 4: Construya el nodo de respuesta LLM
El nodo de respuesta combina los resultados de la búsqueda, el contexto de la memoria y el mensaje del usuario para generar una respuesta fundamentada y consciente del contexto.
from langchain_openai import ChatOpenAI
from langchain_core.messages import AIMessage, SystemMessage
llm = ChatOpenAI(model='gpt-4o', temperature=0)
def respond_node(state: AgentState) -> dict:
system = f"""You are a helpful assistant with web search and memory.
Memory from past conversations:
{state.get('memory_context', 'No prior context.')}
Current search results:
{state.get('search_results', 'No search performed.')}
Use search results for factual claims. Use memory to personalize responses."""
messages = [SystemMessage(content=system)] + list(state['messages'])
response = llm.invoke(messages)
return {'messages': [response]}Paso 5: Armar el gráfico con enrutamiento condicional
Conecte los nodos a un LangGraph con un enrutador que decida si buscar o responder directamente. Las preguntas sobre acontecimientos actuales desencadenan la búsqueda; los seguimientos utilizan la memoria.
from langgraph.graph import StateGraph, END
def should_search(state: AgentState) -> str:
last_msg = state['messages'][-1].content.lower()
search_triggers = ['latest', 'current', 'price', '2026', 'today', 'news', 'best']
if any(t in last_msg for t in search_triggers):
return 'search'
return 'respond'
graph = StateGraph(AgentState)
graph.add_node('search', search_node)
graph.add_node('respond', respond_node)
graph.add_node('memory', memory_node)
graph.set_entry_point('search')
graph.add_conditional_edges('search', should_search, {
'search': 'search',
'respond': 'respond'
})
graph.add_edge('search', 'respond')
graph.add_edge('respond', 'memory')
graph.add_edge('memory', END)
app = graph.compile()
print('Agent graph compiled with search + memory nodes')Ejemplo en Python
import os, requests
from typing import TypedDict, Annotated, Sequence
from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
import operator
API_KEY = os.environ['SCAVIO_API_KEY']
llm = ChatOpenAI(model='gpt-4o', temperature=0)
class AgentState(TypedDict):
messages: Annotated[Sequence[BaseMessage], operator.add]
search_results: str
def search_node(state):
last = state['messages'][-1].content
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': API_KEY, 'Content-Type': 'application/json'},
json={'query': last, 'country_code': 'us'})
results = resp.json().get('organic_results', [])[:5]
ctx = '\n'.join(f'- {r["title"]}: {r.get("snippet", "")}' for r in results)
return {'search_results': ctx}
def respond_node(state):
sys = f'Use these search results:\n{state.get("search_results", "")}'
msgs = [SystemMessage(content=sys)] + list(state['messages'])
return {'messages': [llm.invoke(msgs)]}
graph = StateGraph(AgentState)
graph.add_node('search', search_node)
graph.add_node('respond', respond_node)
graph.set_entry_point('search')
graph.add_edge('search', 'respond')
graph.add_edge('respond', END)
app = graph.compile()
result = app.invoke({'messages': [HumanMessage(content='Best Python frameworks 2026')], 'search_results': ''})
print(result['messages'][-1].content)Ejemplo en JavaScript
// LangGraph is Python-only; this JS example shows the equivalent pattern
const API_KEY = process.env.SCAVIO_API_KEY;
async function searchGrounding(query) {
const resp = await fetch('https://api.scavio.dev/api/v1/search', {
method: 'POST',
headers: { 'x-api-key': API_KEY, 'Content-Type': 'application/json' },
body: JSON.stringify({ query, country_code: 'us' })
});
const data = await resp.json();
return (data.organic_results || []).slice(0, 5)
.map(r => `- ${r.title}: ${r.snippet || ''}`).join('\n');
}
const memory = [];
async function agentTurn(userMessage) {
const context = await searchGrounding(userMessage);
const memoryCtx = memory.slice(-3).join('\n');
const prompt = `Memory:\n${memoryCtx}\n\nSearch:\n${context}\n\nQ: ${userMessage}`;
// Pass prompt to your LLM of choice
console.log(prompt);
memory.push(`User asked: ${userMessage}`);
}
agentTurn('Best Python frameworks 2026').catch(console.error);Salida esperada
Agent graph compiled with search + memory nodes
Based on current search results, the best Python frameworks in 2026 are:
1. FastAPI - async-first, ideal for APIs and microservices
2. Django - full-featured for complex web applications
3. Flask - lightweight and flexible for smaller projects
...
Search grounding cost: $0.005 per agent turn