# agente_investigación.py
# Propósito: Un agente de investigación con instrumentación completa de AgentOps.
# Cada sesión se registra, se reproduce y se realiza un seguimiento de los costos en el panel de AgentOps.
#
# Requisitos previos:
# pip install agentops antrópico python-dotenv
#
# Variables de entorno requeridas (en .env):
# AGENTOPS_API_KEY — de https://app.agentops.ai
# ANTHROPIC_API_KEY — de https://console.anthropic.com
#
# Cómo ejecutar:
# python agente_investigación.py
importar sistema operativo
importar json
tiempo de importación
desde dotenv importar load_dotenv
importar antrópico
agentes de importación
de agentes . SDK . decoradores importan record_function
cargar_dotenv ( )
# ── Inicializar AgentOps ──────────────────────────── ────────────────────────────
# Esto debe llamarse antes de ejecutar cualquier código de agente.
# Las etiquetas le permiten filtrar y agrupar sesiones en el panel.
# El SDK intercepta automáticamente las llamadas LLM una vez inicializadas.
agentes . inicio (
api_key = sistema operativo . entorno [ "AGENTOPS_API_KEY" ] ,
etiquetas = [ "agente-de-investigación" , "producción" , "v1.0" ] ,
auto_start_session = Verdadero # Inicia automáticamente una sesión al iniciar
)
# Inicialice el cliente Anthropic después de AgentOps: el SDK envuelve los clientes LLM
# para capturar automáticamente la entrada, salida, tokens y costo de cada llamada.
cliente = antrópico . Antrópico ( api_key = os.environ [ " ANTHROPIC_API_KEY " ] )
MODELO = "claude-soneto-4-20250514"
# ── Aviso del sistema ────────────────────────────── ───────────────────────────────
# Almacenado como una constante, no en línea: versión controlable y comprobable.
SISTEMA_PROMPT = "" "Eres asistente de investigación. Cuando se te asigna un tema:
1. Utilizar las herramientas disponibles para recopilar información de forma sistemática.
2. Llame a search_topic para obtener una descripción general del tema.
3. Llame a get_key_facts para extraer los puntos más importantes.
4. Llame a format_summary para estructurar el resultado final.
Sea minucioso pero conciso. Siempre llame a format_summary como su paso final." ""
# ── Definiciones de herramientas ───────────────────────────── ─────────────────────────────
# Estas son las herramientas a las que el agente puede recurrir. En un sistema real, search_topic
# llamaría a una API de búsqueda real (Tavily, SerpAPI, etc.). Aqui estan los talones
# que devuelven datos realistas para que puedas ejecutar el ejemplo sin API externas.
HERRAMIENTAS = [
{
"nombre" : "tema_búsqueda" ,
"descripción" : (
"Buscar información completa sobre un tema."
"Devuelve una descripción general con temas clave y contexto".
"Utilice esto como primer paso para cualquier tarea de investigación".
) ,
"esquema_entrada" : {
"tipo" : "objeto" ,
"propiedades" : {
"tema" : {
"tipo" : "cadena" ,
"descripción" : "El tema a investigar. Sea específico".
} ,
"profundidad" : {
"tipo" : "cadena" ,
"enumeración" : [ "descripción general" , "detallado" ] ,
"descripción" : "A qué profundidad buscar. Utilice 'descripción general' primero".
}
} ,
"requerido" : [ "tema" ]
}
} ,
{
"nombre" : "get_key_facts" ,
"descripción" : (
"Extraiga los datos más importantes sobre un tema de los resultados de búsqueda".
"Úselo después de search_topic para identificar los 5-7 puntos más importantes".
) ,
"esquema_entrada" : {
"tipo" : "objeto" ,
"propiedades" : {
"tema" : {
"tipo" : "cadena" ,
"descripción" : "El tema sobre el que extraer datos"
} ,
"enfocar" : {
"tipo" : "cadena" ,
"descripción" : "Opcional: ángulo específico en el que centrarse (por ejemplo, 'desarrollos recientes', 'actores clave')"
}
} ,
"requerido" : [ "tema" ]
}
} ,
{
"nombre" : "formato_summary" ,
"descripción" : (
"Formatee los resultados de la investigación en un resumen estructurado y limpio".
"Llame siempre a esto como el paso final antes de regresar con el usuario".
) ,
"esquema_entrada" : {
"tipo" : "objeto" ,
"propiedades" : {
"título" : {
"tipo" : "cadena" ,
"descripción" : "Título del resumen"
} ,
"puntos_clave" : {
"tipo" : "matriz" ,
"elementos" : { "tipo" : "cadena" } ,
"descripción" : "Lista de hallazgos clave (5-7 elementos)"
} ,
"conclusión" : {
"tipo" : "cadena" ,
"descripción" : "Una síntesis de la investigación en 2 o 3 frases"
}
} ,
"requerido" : [ "título" , "puntos_clave" , "conclusión" ]
}
}
]
# ── Implementaciones de herramientas ─────────────────────────── ───────────────────────────
# @record_function decora cada herramienta para que AgentOps capture:
# – El nombre de la función
# – Argumentos de entrada
# – Valor de retorno
# – tiempo de ejecución
# – Cualquier excepción
# Estos aparecen como intervalos etiquetados en la línea de tiempo de reproducción de la sesión.
@ función_registro ( "tema_búsqueda" )
def tema_búsqueda ( tema : cadena , profundidad : cadena = "descripción general" ) -> dictar :
"" "
Buscar información sobre un tema.
En producción: reemplace este código auxiliar con una llamada API de búsqueda real.
" ""
# Simular la latencia de búsqueda: eliminar en producción
tiempo . dormir ( 0.3 )
# Respuesta de código auxiliar: reemplazar con: tavily_client.search(query=topic)
devolver {
"tema" : tema ,
"profundidad" : profundidad ,
"resultados" : f "Descripción general completa de {tema}: este es un campo en rápida evolución "
f "con novedades significativas en 2025-2026. Los temas clave incluyen "
f "innovación técnica, patrones de adopción e impacto organizacional".
f "Múltiples grupos de investigación y empresas están avanzando activamente en este campo". ,
"recuento_fuente" : 12 ,
"marca de tiempo" : "2026-05-26"
}
@ record_function ( "get_key_facts" )
def get_key_facts ( tema : cadena , enfocar : cadena = Ninguno ) -> dictar :
"" "
Extraer datos clave sobre un tema.
En producción: esto procesaría resultados de búsqueda reales.
" ""
tiempo . dormir ( 0.2 )
nota_enfoque = f "(enfoque: {enfoque})" si concentrarse en otra cosa ""
devolver {
"tema" : tema ,
"enfocar" : nota_enfoque ,
"hechos" : [
f "{topic} ha experimentado un crecimiento interanual del 42 % en la adopción" ,
f "Las organizaciones líderes informan mejoras de productividad entre 3 y 5 veces mayores" ,
f "Los desafíos técnicos clave incluyen la confiabilidad, el costo y la gobernanza" ,
f "Se prevé que el mercado alcance los 4.900 millones de dólares en 2028" ,
f "Las herramientas de código abierto han madurado significativamente en los últimos 18 meses" ,
] ,
"confianza" : "alto"
}
@ record_function ( "formato_summary" )
def format_summary ( título : cadena , puntos_clave : lista , conclusión : cadena ) -> dictar :
"" "
Formatee la investigación en un resumen estructurado.
Este es siempre el paso final en el flujo de trabajo de la investigación.
" ""
devolver {
"título" : título ,
"puntos_clave" : puntos_clave ,
"conclusión" : conclusión ,
"formato" : "resumen_estructurado" ,
"generado_at" : "2026-05-26"
}
def ejecutar_herramienta ( nombre_herramienta : cadena , entrada_herramienta : dictado ) -> cadena :
"" "
Enrute las llamadas a la herramienta hacia la implementación correcta.
Devuelve el resultado como una cadena JSON para que lo lea el modelo.
" ""
si nombre_herramienta == "tema_búsqueda" :
resultado = tema_búsqueda ( * * entrada_herramienta )
elif nombre_herramienta == "get_key_facts" :
resultado = get_key_facts ( * * entrada_herramienta )
elif nombre_herramienta == "formato_summary" :
resultado = formato_summary ( * * entrada_herramienta )
demás :
resultado = { "error" : f "Herramienta desconocida: {tool_name}" }
devolver json . volcados ( resultado )
# ── El bucle del agente ────────────────────────────── ───────────────────────────────
def run_research_agent ( tema : cadena ) -> dictar :
"" "
Ejecute el agente de investigación sobre un tema determinado.
El bucle:
1. Envía el gol a Claude con las herramientas disponibles.
2. Si Claude quiere llamar a una herramienta, ejecútela y devuelva el resultado.
3. Continúe hasta que Claude indique que ha terminado (stop_reason == 'end_turn')
4. Devolver el resumen estructurado final.
AgentOps captura cada iteración automáticamente porque:
– El cliente LLM se empaqueta después de agentops.init()
– Cada herramienta está decorada con @record_function
– La sesión abarca todo el ciclo de vida desde init hasta end_session()
" ""
print ( f "nIniciando agente de investigación para el tema: '{topic}'" )
print ( "La sesión será visible en https://app.agentops.ain" )
mensajes = [
{ "role" : "usuario" , "contenido" : f "Investigue este tema y produzca un resumen estructurado: {tema}" }
]
resumen_final = Ninguno
iteración = 0
iteraciones_max = 10 # Límite de seguridad: evita bucles fuera de control
mientras iteración < iteraciones_max :
iteración += 1
print ( f "Iteración {iteración}: Llamando a Claude…" )
respuesta = cliente . mensajes . crear (
modelo = MODELO ,
tokens_max = 4096 ,
sistema = SISTEMA_PROMPT ,
herramientas = HERRAMIENTAS ,
mensajes = mensajes
)
imprimir ( f " stop_reason: {respuesta.stop_reason}" )
# Agregar respuesta del asistente al historial de mensajes
mensajes . agregar ( { "rol" : "asistente" , "contenido" : respuesta . contenido } )
# Si Claude termina, extrae el resumen final y sal.
si respuesta . motivo_parada == "end_turn" :
# Busque el resultado format_summary en el historial de mensajes
para mensaje en invertido ( mensajes ) :
si mensaje [ "rol" ] == "usuario" y isinstance ( msg [ "contenido" ] , lista ) :
para bosquejar mensaje [ "contenido" ] :
si ( hasattr ( bloque , "tipo" ) y bloquear . tipo == "resultado_herramienta" ) :
intentar :
datos_resultados = json . cargas ( bloque . contenido [ 0 ] . texto )
si datos_resultados . obtener ( "formato" ) == "resumen_estructurado" :
resumen_final = datos_resultados
romper
excepto ( json . JSONDecodeError , ( Error de atributo , error clave , Error de índice , Error de tipo ) ) :
aprobar
si resumen_final :
romper
romper
# Procesar llamadas a herramientas si Claude quiere usar herramientas
si respuesta . motivo_parada == "uso_herramienta" :
resultados_herramienta = [ ]
para bosquejar respuesta . contenido :
si bloquear . tipo == "uso_herramienta" :
print ( f " Llamada a herramienta: {block.name}({json.dumps(block.input, indent=2)})" )
resultado = ejecutar_tool ( bloque . nombre , bloquear . aporte )
imprimir ( f " Resultado: {resultado[:100]}…" )
resultados_herramienta . agregar ( {
"tipo" : "resultado_herramienta" ,
"tool_use_id" : bloquear . identificación ,
"contenido" : resultado
} )
# Devolver los resultados de la herramienta a Claude
mensajes . agregar ( { "rol" : "usuario" , "contenido" : resultados_herramienta } )
si iteración >= iteraciones_max :
print ( f "ADVERTENCIA: El agente alcanzó el número máximo de iteraciones ({max_iterations}). Posible bucle detectado". )
# AgentOps mostrará esto como una sesión que finaliza en Error
agentes . end_session ( "falla" )
devolver { "error" : "Se alcanzó el máximo de iteraciones: verifique la repetición de la sesión para analizar el bucle" }
# Finalizar sesión con éxito: esto finaliza la sesión en AgentOps
# La repetición de la sesión ya está disponible en app.agentops.ai
agentes . end_session ( "Éxito" )
devolver resumen_final o { "mensaje" : "Investigación completa: verifique la repetición de la sesión para ver el seguimiento completo" }
# ── Ejecutar el agente ────────────────────────────── ───────────────────────────────
si __nombre__ == "__principal__" :
tema = "AgentOps y observabilidad de agentes de IA en 2026"
intentar :
resultado = run_research_agent ( tema )
imprimir ( "n" + "=" * 60 )
imprimir ( "RESUMEN DE LA INVESTIGACIÓN" )
imprimir ( "=" * 60 )
si "error" en resultado :
imprimir ( f "Error: {resultado['error']}" )
demás :
print ( f "Título: {result.get('título', 'N/A')}" )
imprimir ( "nPuntos clave: " )
para i , señalar en enumerar ( resultado . obtener ( "puntos_clave" , [ ] ) , 1 ) :
imprimir ( f " {i}. {punto}" )
print ( f "nConclusión: {result.get('conclusión', 'N/A')}" )
imprimir ( "n" + "=" * 60 )
print ( "Repetición de la sesión disponible en: https://app.agentops.ai" )
print ( "Busque su sesión etiquetada como 'agente-de-investigación'" )
imprimir ( "=" * 60 )
excepto KeyboardInterrupt :
# La sesión limpia finaliza si el usuario interrumpe
agentes . end_session ( "falla" )
print ( "nSesión finalizada por el usuario. Seguimiento parcial guardado en AgentOps." )
excepto excepción como mi :
# Registrar fallas para que aparezcan en el tablero
agentes . end_session ( "falla" )
print ( f "Agente fallido: {e}" )
aumentar