Agregar una herramienta de búsqueda a un agente de investigación LangGraph significa crear un nodo de herramienta en el gráfico de estado que llama a una API de búsqueda y dirige los resultados de regreso al ciclo de razonamiento del agente. LangGraph modela los flujos de trabajo de los agentes como máquinas de estado donde los nodos realizan acciones y los bordes definen el flujo de control. Al agregar un nodo de búsqueda que llama a la API de Scavio y actualiza el estado del agente con resultados estructurados, le brinda al agente de investigación la capacidad de recopilar evidencia en vivo, verificar afirmaciones y descubrir fuentes dentro de su ciclo de ejecución de gráficos.
Requisitos previos
- Python 3.10+
- langgraph y langchain-core instalados
- Clave API de Scavio de scavio.dev
- Comprensión básica de las máquinas de estados LangGraph
Guia paso a paso
Paso 1: Definir el estado del agente
Cree un TypedDict que contenga los mensajes de conversación, los resultados de la búsqueda y el estado de la investigación para la máquina de estado LangGraph.
Python
from typing import TypedDict, Annotated, Sequence
from langchain_core.messages import BaseMessage
import operator
class ResearchState(TypedDict):
messages: Annotated[Sequence[BaseMessage], operator.add]
search_results: list[dict]
search_query: str
research_complete: boolPaso 2: Crear el nodo de la herramienta de búsqueda
Cree una función de nodo que extraiga la consulta de búsqueda del estado, llame a la API de Scavio y actualice el estado con los resultados.
Python
import os, requests
from langchain_core.messages import ToolMessage
H = {'x-api-key': os.environ['SCAVIO_API_KEY'], 'Content-Type': 'application/json'}
def search_node(state: ResearchState) -> dict:
query = state['search_query']
data = requests.post('https://api.scavio.dev/api/v1/search',
headers=H, json={'query': query, 'country_code': 'us'}).json()
results = [{
'title': r.get('title', ''),
'url': r.get('link', ''),
'snippet': r.get('snippet', ''),
} for r in data.get('organic_results', [])[:5]]
summary = '\n'.join([f"- {r['title']}: {r['snippet']}" for r in results])
return {
'search_results': results,
'messages': [ToolMessage(content=f'Search results for "{query}":\n{summary}',
tool_call_id='search')],
}Paso 3: Construya el nodo de razonamiento y el enrutador
Cree el nodo de razonamiento del agente que decida si buscar, continuar investigando o finalizar la respuesta.
Python
from langchain_openai import ChatOpenAI
from langchain_core.messages import AIMessage, HumanMessage
llm = ChatOpenAI(model='gpt-4o')
def reasoning_node(state: ResearchState) -> dict:
system = ('You are a research agent. Analyze the conversation and decide:\n'
'1. If you need more data, respond with SEARCH: <query>\n'
'2. If you have enough data, respond with ANSWER: <your answer>')
messages = [HumanMessage(content=system)] + list(state['messages'])
response = llm.invoke(messages)
content = response.content
if content.startswith('SEARCH:'):
query = content.replace('SEARCH:', '').strip()
return {'messages': [response], 'search_query': query, 'research_complete': False}
else:
return {'messages': [response], 'research_complete': True}
def should_search(state: ResearchState) -> str:
if state.get('research_complete'):
return 'done'
return 'search'Paso 4: Armar y ejecutar el gráfico
Conecte los nodos en una máquina de estado LangGraph con bordes condicionales y ejecute una consulta de investigación.
Python
from langgraph.graph import StateGraph, END
def build_research_graph():
graph = StateGraph(ResearchState)
graph.add_node('reason', reasoning_node)
graph.add_node('search', search_node)
graph.add_conditional_edges('reason', should_search, {
'search': 'search',
'done': END,
})
graph.add_edge('search', 'reason')
graph.set_entry_point('reason')
return graph.compile()
research_agent = build_research_graph()
# Run a research task
result = research_agent.invoke({
'messages': [HumanMessage(content='What are the best vector databases for RAG in 2026?')],
'search_results': [],
'search_query': '',
'research_complete': False,
})
final_message = result['messages'][-1].content
print(final_message)Ejemplo en Python
Python
import os, requests, operator
from typing import TypedDict, Annotated, Sequence
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
H = {'x-api-key': os.environ['SCAVIO_API_KEY'], 'Content-Type': 'application/json'}
llm = ChatOpenAI(model='gpt-4o')
class ResearchState(TypedDict):
messages: Annotated[Sequence[BaseMessage], operator.add]
search_results: list[dict]
search_query: str
research_complete: bool
def search_node(state: ResearchState) -> dict:
q = state['search_query']
data = requests.post('https://api.scavio.dev/api/v1/search',
headers=H, json={'query': q, 'country_code': 'us'}).json()
results = [{'title': r.get('title',''), 'url': r.get('link',''),
'snippet': r.get('snippet','')} for r in data.get('organic_results',[])[:5]]
summary = '\n'.join([f"- {r['title']}: {r['snippet']}" for r in results])
return {'search_results': results,
'messages': [ToolMessage(content=f'Results for "{q}":\n{summary}', tool_call_id='search')]}
def reason_node(state: ResearchState) -> dict:
prompt = ('Research agent. Need more data? Say SEARCH: <query>. '
'Have enough? Say ANSWER: <response>')
msgs = [HumanMessage(content=prompt)] + list(state['messages'])
resp = llm.invoke(msgs)
if resp.content.startswith('SEARCH:'):
return {'messages': [resp], 'search_query': resp.content[7:].strip(),
'research_complete': False}
return {'messages': [resp], 'research_complete': True}
def router(state: ResearchState) -> str:
return 'done' if state.get('research_complete') else 'search'
graph = StateGraph(ResearchState)
graph.add_node('reason', reason_node)
graph.add_node('search', search_node)
graph.add_conditional_edges('reason', router, {'search': 'search', 'done': END})
graph.add_edge('search', 'reason')
graph.set_entry_point('reason')
agent = graph.compile()
result = agent.invoke({
'messages': [HumanMessage(content='Best vector databases for RAG in 2026?')],
'search_results': [], 'search_query': '', 'research_complete': False})
print(result['messages'][-1].content)Ejemplo en JavaScript
JavaScript
// LangGraph.js research agent with Scavio search
const H = {'x-api-key': process.env.SCAVIO_API_KEY, 'Content-Type': 'application/json'};
async function searchNode(state) {
const data = await fetch('https://api.scavio.dev/api/v1/search', {
method: 'POST', headers: H,
body: JSON.stringify({query: state.searchQuery, country_code: 'us'})
}).then(r => r.json());
const results = (data.organic_results || []).slice(0, 5).map(r => ({
title: r.title || '', url: r.link || '', snippet: r.snippet || ''
}));
const summary = results.map(r => \`- \${r.title}: \${r.snippet}\`).join('\n');
return {
searchResults: results,
messages: [...state.messages, {role: 'tool', content: \`Results: \${summary}\`}],
};
}
// Reasoning node calls your LLM to decide search vs answer
async function reasonNode(state) {
// Call LLM with state.messages, parse SEARCH:/ANSWER: prefix
// Return updated state with searchQuery or researchComplete
console.log(\`Processing \${state.messages.length} messages\`);
return state;
}
// Wire into LangGraph.js StateGraph
// const graph = new StateGraph({channels: {...}})
// graph.addNode('reason', reasonNode)
// graph.addNode('search', searchNode)
// graph.addConditionalEdges('reason', router, {search: 'search', done: END})
console.log('LangGraph research agent with search ready');Salida esperada
JSON
Research agent execution:
reason -> SEARCH: vector databases RAG comparison 2026
search -> 5 results from Scavio
reason -> SEARCH: pinecone vs weaviate vs qdrant benchmarks
search -> 5 results from Scavio
reason -> ANSWER: The top vector databases for RAG in 2026 are...
Final answer includes cited sources from live search data.