Diseño de bucles de investigación en varios pasos
Planificar → buscar → leer → extraer → sintetizar → repetir hasta alcanzar una profundidad suficiente.
Diseño de bucles de investigación en varios pasos es una lección gratuita de AI Agents en CoddyKit. Esta es la lección 1 de 4. Puedes leer la lección completa abajo gratuitamente — luego la practicas en el navegador con un editor de código integrado y un tutor de IA 24/7. Forma parte de la ruta de aprendizaje de AI Agents, y tu progreso se sincroniza en la web y la app de CoddyKit. El curso de AI Agents incluye 4 lecciones en total.
¿Qué es un bucle de investigación?
Un bucle de investigación es un patrón agéntico en el que un LLM planifica, busca, lee, extrae información, identifica lagunas y repite el proceso hasta alcanzar el objetivo de investigación.
A diferencia de una búsqueda única, el bucle se adapta según lo que encuentra: sigue pistas inesperadas y descarta callejones sin salida.
Fase 1: Descomposición de la pregunta
El primer paso consiste en dividir la pregunta de investigación en subpreguntas. Esto crea un plan de búsqueda dirigido y evita que el agente navegue sin rumbo.
import openai, json
client = openai.OpenAI(api_key='YOUR_OPENAI_KEY')
def decompose_question(question: str) -> list[str]:
prompt = (
f'Break this research question into 3-5 focused sub-questions.\n'
f'Each sub-question should be independently searchable.\n'
f'Question: "{question}"\n'
f'Return JSON: {{"sub_questions": ["..."]}}'
)
resp = client.chat.completions.create(
model='gpt-4o',
messages=[{'role': 'user', 'content': prompt}],
response_format={'type': 'json_object'}
)
return json.loads(resp.choices[0].message.content)['sub_questions']
questions = decompose_question('What are the main causes of inflation in 2024?')
print(questions)Fase 2: Búsqueda web
Realice una búsqueda web para cada subpregunta. Use una API de búsqueda, como Serper, Brave o Bing, para obtener una lista de URL y fragmentos. No lea todas las páginas: dé prioridad primero a las fuentes de alta calidad.
import requests
SERPER_KEY = 'YOUR_SERPER_API_KEY'
def search_web(query: str, num_results: int = 5) -> list[dict]:
resp = requests.post(
'https://google.serper.dev/search',
headers={'X-API-KEY': SERPER_KEY, 'Content-Type': 'application/json'},
json={'q': query, 'num': num_results}
)
resp.raise_for_status()
results = resp.json().get('organic', [])
return [
{'title': r['title'], 'url': r['link'], 'snippet': r.get('snippet', '')}
for r in results
]Fase 3: Lectura y extracción de datos
Obtenga cada URL y extraiga los datos clave relevantes para la subpregunta. Use un LLM para leer el artículo y generar una lista de datos junto con la URL de origen.
import httpx
from bs4 import BeautifulSoup
def fetch_text(url: str, max_chars: int = 4000) -> str:
try:
resp = httpx.get(url, timeout=10, follow_redirects=True,
headers={'User-Agent': 'ResearchAgent/1.0'})
soup = BeautifulSoup(resp.text, 'html.parser')
for tag in soup(['script', 'style', 'nav', 'footer']):
tag.decompose()
return soup.get_text(separator=' ', strip=True)[:max_chars]
except Exception:
return ''
def extract_facts(text: str, question: str, url: str) -> list[dict]:
prompt = (
f'Extract key facts from the text that answer: "{question}"\n'
f'Return JSON: {{"facts": ["fact1", ...]}}\n\n'
f'TEXT:\n{text[:3000]}'
)
resp = client.chat.completions.create(
model='gpt-4o-mini',
messages=[{'role': 'user', 'content': prompt}],
response_format={'type': 'json_object'}
)
facts = json.loads(resp.choices[0].message.content).get('facts', [])
return [{'fact': f, 'source': url} for f in facts]Fase 4: Identificación de lagunas de conocimiento
Después de leer, pida al LLM que revise los datos acumulados e identifique qué sigue sin conocerse. Estas lagunas se convierten en las consultas de búsqueda de la siguiente ronda.
def identify_gaps(original_question: str, facts: list[dict]) -> list[str]:
fact_text = '\n'.join(f'- {f["fact"]}' for f in facts[:20])
prompt = (
f'Original question: "{original_question}"\n'
f'Facts gathered so far:\n{fact_text}\n\n'
f'What key aspects are still unanswered? '
f'Return 0-3 follow-up search queries (0 if research is complete).\n'
f'JSON: {{"gaps": ["search query 1", ...]}}'
)
resp = client.chat.completions.create(
model='gpt-4o',
messages=[{'role': 'user', 'content': prompt}],
response_format={'type': 'json_object'}
)
return json.loads(resp.choices[0].message.content).get('gaps', [])Condiciones de finalización
El bucle necesita una condición de parada clara para evitar ejecutarse indefinidamente. Deténgalo cuando:
- No se identifiquen nuevas lagunas
- No se hayan añadido nuevos datos en la última iteración
- Se alcance el número máximo de iteraciones (límite de seguridad)
- El total de datos supere un umbral (profundidad suficiente)
def should_continue(gaps: list[str], new_facts_this_round: int,
iteration: int, total_facts: int) -> bool:
if iteration >= 5:
return False # Hard cap: 5 iterations
if new_facts_this_round == 0:
return False # No new information found
if not gaps:
return False # LLM says research is complete
if total_facts >= 50:
return False # Sufficient depth reached
return True
if __name__ == '__main__':
print('Continue (has gaps, new facts)?', should_continue(['gap1'], 4, iteration=1, total_facts=10))
print('Continue (no new facts)?', should_continue(['gap1'], 0, iteration=1, total_facts=10))
Eliminación de datos duplicados
Varias fuentes suelen informar del mismo dato. Elimine los duplicados pidiendo al LLM que fusione los datos semánticamente equivalentes y conserve la versión respaldada por las mejores fuentes.
def deduplicate_facts(facts: list[dict]) -> list[dict]:
if len(facts) <= 3:
return facts
fact_text = '\n'.join(
f'{i}: {f["fact"]} (source: {f["source"]})' for i, f in enumerate(facts)
)
prompt = (
f'Remove duplicate or near-duplicate facts. Keep the most informative version.\n'
f'Return JSON: {{"keep_indices": [0, 1, ...]}}\n\n'
f'FACTS:\n{fact_text}'
)
resp = client.chat.completions.create(
model='gpt-4o-mini',
messages=[{'role': 'user', 'content': prompt}],
response_format={'type': 'json_object'}
)
indices = json.loads(resp.choices[0].message.content).get('keep_indices', [])
return [facts[i] for i in indices if i < len(facts)]El bucle de investigación completo
Combine todas las fases en una única función research() que ejecute el bucle hasta su finalización.
def research(question: str) -> dict:
all_facts = []
iteration = 0
# Phase 1: Decompose into sub-questions
queries = decompose_question(question)
while True:
new_facts_this_round = 0
for query in queries:
results = search_web(query, num_results=3)
for result in results:
text = fetch_text(result['url'])
if not text:
continue
facts = extract_facts(text, query, result['url'])
all_facts.extend(facts)
new_facts_this_round += len(facts)
all_facts = deduplicate_facts(all_facts)
gaps = identify_gaps(question, all_facts)
iteration += 1
if not should_continue(gaps, new_facts_this_round, iteration, len(all_facts)):
break
queries = gaps # next iteration searches for the gaps
return {'facts': all_facts, 'iterations': iteration}Seguimiento de la procedencia
Cada dato debe incluir su URL de origen para que el informe final pueda incorporar citas. Nunca elimine los metadatos de origen durante el procesamiento: son necesarios para la capa de citas.
def add_fact(fact_list: list, fact_text: str, source_url: str, iteration: int):
fact_list.append({
'fact': fact_text,
'source': source_url,
'iteration': iteration,
'verified': False # set to True after cross-referencing
})
# Example:
facts_store = []
add_fact(facts_store, 'Global inflation peaked at 9.1% in June 2022',
'https://bls.gov/news.release/cpi.htm', iteration=1)
print(f'Logged {len(facts_store)} fact(s):')
for f in facts_store:
print(f" - {f['fact']} (source: {f['source']})")
Búsqueda en paralelo para aumentar la velocidad
Las búsquedas secuenciales son lentas: 5 consultas × 3 URL × 1 obtención cada una = 15 llamadas HTTP secuenciales. Use concurrent.futures para paralelizar las obtenciones dentro de cada iteración.
from concurrent.futures import ThreadPoolExecutor, as_completed
def parallel_research_round(queries: list[str]) -> list[dict]:
collected = []
def process_query(query):
results = search_web(query, num_results=3)
facts = []
for r in results:
text = fetch_text(r['url'])
if text:
facts.extend(extract_facts(text, query, r['url']))
return facts
with ThreadPoolExecutor(max_workers=4) as ex:
futures = {ex.submit(process_query, q): q for q in queries}
for future in as_completed(futures):
collected.extend(future.result())
return collectedSupervisión del progreso del bucle
Registre cada iteración para poder depurar por qué se detuvo el bucle o por qué se ejecutó durante más tiempo del esperado. Incluya el número de consultas, los datos añadidos y las lagunas identificadas.
import logging
log = logging.getLogger('research_loop')
import sys
logging.basicConfig(level=logging.INFO, stream=sys.stdout)
def log_iteration(iteration: int, queries: list[str],
new_facts: int, total_facts: int, gaps: list[str]):
log.info(
'Iteration %d | Queries: %d | New facts: %d | Total: %d | Gaps: %d',
iteration, len(queries), new_facts, total_facts, len(gaps)
)
if gaps:
for g in gaps:
log.debug(' Gap query: %s', g)
if __name__ == '__main__':
log_iteration(iteration=2, queries=['who are our top competitors?'], new_facts=4, total_facts=12, gaps=['pricing data'])
¿Qué termina el bucle de investigación cuando no se encuentra información nueva?
Comprender las condiciones de finalización evita los bucles infinitos y garantiza que el agente entregue resultados dentro de un plazo razonable.
Resumen del diseño del bucle de investigación
El bucle de investigación de varios pasos sigue esta secuencia: descomponer → buscar → leer → extraer → identificar lagunas → repetir. Deténgalo cuando no queden lagunas, no se encuentren datos nuevos o se alcance un límite estricto de iteraciones.
Realice siempre un seguimiento de las URL de origen de cada dato, paralelice las búsquedas para aumentar la velocidad y elimine los datos duplicados antes de sintetizarlos.
Preguntas frecuentes
¿La lección «Diseño de bucles de investigación en varios pasos» es gratis?
Sí — el texto completo de «Diseño de bucles de investigación en varios pasos» es gratis para leer aquí en la web. Para practicarla de forma interactiva (editor de código integrado y tutor de IA 24/7) y desbloquear el resto del curso de AI Agents, actualiza a CoddyKit PRO. El curso de AI Agents incluye 4 lecciones en total.
¿Qué aprenderé en «Diseño de bucles de investigación en varios pasos»?
Planificar → buscar → leer → extraer → sintetizar → repetir hasta alcanzar una profundidad suficiente. Practicas AI Agents con código real que ejecutas directamente en el navegador, y un tutor de IA 24/7 responde tus preguntas mientras trabajas en la lección.
¿Necesito experiencia previa para empezar AI Agents?
No se requiere experiencia previa. AI Agents en CoddyKit está estructurado para principiantes hasta estudiantes avanzados, así que puedes empezar aquí o desde el inicio y avanzar a tu ritmo. Esta es la lección 1 de 4.
¿Cuánto tiempo toma la lección «Diseño de bucles de investigación en varios pasos»?
La mayoría de las lecciones de CoddyKit toman alrededor de 5–10 minutos. Cada una es compacta e interactiva, así que avanzas constantemente y retomas exactamente por donde dejaste en la web y la app.
¿Puedo escribir y ejecutar código en esta lección de AI Agents?
Sí. Cada lección de AI Agents incluye un editor de código integrado, así que escribes y ejecutas código real directamente en tu navegador y obtienes retroalimentación instantánea de IA — sin configuración local necesaria.
Todas las lecciones de este curso
- Diseño de bucles de investigación en varios pasos
- Verificación de fuentes y citas
- Generación de informes estructurados
- Comprobación de datos y prevención de alucinaciones