En este tutorial, creamos un canal de aprendizaje avanzado de extremo a extremo sobre Atomic-Agents conectando interfaces de agentes tipificadas, indicaciones estructuradas y una capa de recuperación compacta que basa los resultados en documentación real del proyecto. Además, demostramos cómo planificar la recuperación, recuperar contexto relevante, inyectarlo dinámicamente en un agente de respuesta y ejecutar un bucle interactivo que convierte la configuración en un asistente de investigación reutilizable para cualquier pregunta nueva sobre agentes atómicos. Consulta los CÓDIGOS COMPLETOS aquí.
“atomic-agents”, “instructor”, “openai”, “pydantic”,
“requests”, “beautifulsoup4”, “scikit-learn”]) de getpass importar getpass si no os.environ.get(“OPENAI_API_KEY”): os.environ[“OPENAI_API_KEY”] = getpass(“Ingrese OPENAI_API_KEY (entrada oculta): “).strip() MODELO = os.environ.get(“OPENAI_MODEL”, “gpt-4o-mini”) de pydantic import Campo de openai import OpenAI import instructor de atomic_agents import AtomicAgent, AgentConfig, BaseIOSchema de atomic_agents.context import SystemPromptGenerator, ChatHistory, Solicitudes de importación BaseDynamicContextProvider desde bs4 import BeautifulSoup
Instalamos todos los paquetes necesarios, importamos las primitivas principales de Atomic-Agents y configuramos dependencias compatibles con Colab en un solo lugar. Capturamos de forma segura la clave API de OpenAI del teclado y la almacenamos en el entorno para que el código posterior nunca codifique secretos. También fijamos un nombre de modelo predeterminado y lo mantenemos configurable a través de una variable de entorno.
trozos = []
i = 0 mientras i < len(texto): fragmento = texto[i:i+max_chars].strip() if trozo: trozos.append(chunk) i += max_chars - superponer devolver trozos def abrazadera(s: str, n: int = 800) -> str: s = (s o “”).strip() devolver s si len(s) <= n else s[:n].rstrip() + "..."
Recuperamos páginas web del repositorio y los documentos de Atomic Agents, luego las limpiamos y las convertimos en texto sin formato para que la recuperación sea confiable. Dividimos documentos largos en segmentos superpuestos, preservando el contexto y manteniendo cada fragmento lo suficientemente pequeño para clasificarlos y citarlos. También agregamos un pequeño ayudante para sujetar fragmentos largos para que nuestro contexto inyectado siga siendo legible.
para doc_id, (url, raw) en docs.items(): para idx, ch en enumerate(chunk_text(raw)): self.items.append((doc_id, url, idx, ch)) si no es self.items: rise RuntimeError(“No se recuperaron documentos; no se puede generar el índice TF-IDF.”) self.vectorizer = TfidfVectorizer(stop_words=”inglés”, max_features=50000) self.matrix = self.vectorizer.fit_transform([it[3] para ello en self.items]) def search(self, query: str, k: int = 6) -> Lista[Snippet]: qv = self.vectorizador.transformar([query]) sims = coseno_similaridad(qv, self.matrix).ravel() arriba = sims.argsort()[::-1][:k]
fuera = []
para j en la parte superior: doc_id, url, chunk_id, txt = self.items[j]
out.append(Fragmento(doc_id=doc_id, url=url, chunk_id=chunk_id, text=txt, puntuación=float(sims[j]))) devuelve la clase RetrievedContextProvider(BaseDynamicContextProvider): def __init__(self, título: str, fragmentos: Lista[Snippet]): super().__init__(título=título) self.snippets = fragmentos def get_info(self) -> str: bloques = []
para s en self.snippets: bloques.append( f”[{s.doc_id}#{s.chunk_id}] (score={s.score:.3f}) {s.url}\n{clamp(s.text, 900)}” ) return “\n\n”.join(bloques)
Construimos un mini sistema de recuperación utilizando TF-IDF y similitud de coseno sobre el corpus de documentación fragmentado. Envolvemos cada fragmento recuperado en un objeto fragmento estructurado para realizar un seguimiento de los ID de documentos, los ID de fragmentos y las puntuaciones de citas. Luego inyectamos fragmentos mejor clasificados en el tiempo de ejecución del agente a través de un proveedor de contexto dinámico, manteniendo al agente que responde con los pies en la tierra. Consulta los CÓDIGOS COMPLETOS aquí.
must_cover: Lista[str]
controles_de seguridad: Lista[str]
class AnswerInput(BaseIOSchema): “””Esquema de entrada para el agente de respuesta: pregunta del usuario más restricciones de estilo.””” pregunta: str estilo: str = “conciso pero avanzado” class AnswerOutput(BaseIOSchema): “””Esquema de salida para el agente de respuesta: respuesta fundamentada, próximos pasos y qué citas se utilizaron.””” respuesta: str next_steps: Lista[str]
citas_usadas: Lista[str]
cliente = instructor.from_openai(OpenAI(api_key=os.environ[“OPENAI_API_KEY”])) planificador_prompt = SystemPromptGenerator (fondo =[
“You are a rigorous research planner for a small RAG system.”,
“You propose retrieval queries that are diverse (lexical + semantic) and designed to find authoritative info.”,
“You do NOT answer the task; you only plan retrieval.”
]pasos =[
“Read the task.”,
“Propose diverse retrieval queries (not too long).”,
“List must-cover aspects and safety checks.”
]instrucciones_salida=[
“Return strictly the PlanOutput schema.”,
“Queries must be directly usable as search strings.”,
“Must-cover should be 4–8 bullets.”
]
) planificador = Agente Atómico[PlanInput, PlanOutput]( config=AgentConfig( cliente=cliente, modelo=MODELO, system_prompt_generator=planner_prompt, historial=ChatHistory(), ) ) respuesta_prompt = SystemPromptGenerator( fondo=[
“You are an expert technical tutor for Atomic Agents (atomic-agents).”,
“You are given retrieved context snippets with IDs like [doc#chunk].”, “Debe fundamentar las afirmaciones en los fragmentos proporcionados y citarlos en línea.” ], pasos=[
“Read the question and the provided context.”,
“Synthesize an accurate answer using only supported facts.”,
“Cite claims inline using the provided snippet IDs.”
]instrucciones_salida=[
“Use inline citations like [readme#12] o [docs_home#3].”, “Si el contexto no admite algo, dígalo brevemente y sugiera qué recuperar a continuación.”, “Devuelva estrictamente el esquema AnswerOutput”. ]) contestador = AtomicAgent[AnswerInput, AnswerOutput]( config=AgentConfig( cliente=cliente, modelo=MODELO, system_prompt_generator=answerer_prompt, historial=ChatHistory(), ) )
Definimos esquemas de tipo estricto para las entradas y salidas del planificador y del contestador, e incluimos cadenas de documentación para satisfacer los requisitos del esquema de Atomic Agents. Creamos un cliente OpenAI diseñado por un instructor y configuramos dos agentes atómicos con indicaciones explícitas del sistema e historial de chat. Aplicamos resultados estructurados para que el planificador genere consultas y el que responde produzca una respuesta citada con los próximos pasos claros.
para q en plan.queries: all_snips.extend(retriever.search(q, k=max(2, k // 2))) mejor: Dict[Tuple[str, int]Fragmento]= {} para s en all_snips: clave = (s.doc_id, s.chunk_id) si (la clave no está en mejor) o (s.score > mejor[key].score): mejor[key] = s recortes = ordenado(mejores.valores(), clave=lambda x: x.puntuación, reverso=Verdadero)[:k]
ctx = RetrievedContextProvider(title=”Contexto de agentes atómicos recuperados”, snippets=snips) respuesta Agentes en un nivel avanzado: explique los componentes básicos y muestre cómo encadenar agentes con esquemas escritos y contexto dinámico”. run_atomic_rag(demo_q, k=7, verbose=True) mientras es Verdadero: user_q = input(“\nTu pregunta> “).strip() si no es user_q o user_q.lower() en {“exit”, “quit”}: romper run_atomic_rag(user_q, k=7, verbose=True)
Recuperamos un pequeño conjunto de fuentes autorizadas de Agentes Atómicos y creamos un índice de recuperación local a partir de ellas. Implementamos una función de canalización completa que planifica consultas, recupera contexto relevante, lo inyecta y produce una respuesta final fundamentada. Terminamos ejecutando una consulta de demostración y lanzando un bucle interactivo para que podamos seguir haciendo preguntas y obteniendo respuestas citadas.
En conclusión, completamos el flujo de trabajo de Atomic-Agents en Colab, separando claramente la planificación, la recuperación, la respuesta y garantizando una escritura sólida. Mantuvimos el sistema conectado inyectando solo los fragmentos de documentación de mayor señal como contexto dinámico, y aplicamos una disciplina de citas que hace que los resultados sean auditables. A partir de aquí, podemos escalar este patrón agregando más fuentes, intercambiando recuperadores o reclasificadores más fuertes, introduciendo agentes de uso de herramientas y convirtiendo el proceso en un asistente de investigación de nivel de producción que siga siendo rápido y confiable.
Consulta los CÓDIGOS COMPLETOS aquí. Además, no dude en seguirnos en Twitter y no olvide unirse a nuestro SubReddit de más de 100.000 ML y suscribirse a nuestro boletín. ¡Esperar! estas en telegrama? Ahora también puedes unirte a nosotros en Telegram.