Cómo diseñar una arquitectura de IA agente con LangGraph y OpenAI utilizando deliberación adaptativa, gráficos de memoria y bucles de reflexión

En este tutorial, construimos un sistema de IA Agentic genuinamente avanzado utilizando modelos LangGraph y OpenAI yendo más allá de los simples ciclos planificadores y ejecutores. Implementamos deliberación adaptativa, donde el agente decide dinámicamente entre razonamiento rápido y profundo; un gráfico de memoria agente estilo Zettelkasten que almacena conocimiento atómico y vincula automáticamente experiencias relacionadas; y un mecanismo gobernado de uso de herramientas que impone restricciones durante la ejecución. Al combinar la gestión del estado estructurado, la recuperación consciente de la memoria, el aprendizaje reflexivo y la invocación controlada de herramientas, demostramos cómo los sistemas agentes modernos pueden razonar, actuar, aprender y evolucionar en lugar de responder en una sola pasada. Consulta los CÓDIGOS COMPLETOS aquí.

!pip -q install -U langgraph langchain-openai langchain-core pydantic numpy networkx solicita importar sistema operativo, getpass, json, tiempo, operador desde escribir import List, Dict, Cualquiera, Opcional, Literal desde escribiendo_extensiones importar TypedDict, Importar anotado numpy como np importar networkx como nx desde pydantic import BaseModel, Campo desde langchain_openai importar ChatOpenAI, OpenAIEmbeddings de langchain_core.messages importa SystemMessage, HumanMessage, ToolMessage, AnyMessage de langchain_core.tools importa la herramienta de langgraph.graph importa StateGraph, START, END de langgraph.checkpoint.memory importa InMemorySaver

Configuramos el entorno de ejecución instalando todas las bibliotecas necesarias e importando los módulos principales. Reunimos LangGraph para orquestación, LangChain para abstracciones de modelos y herramientas, y bibliotecas de soporte para gráficos de memoria y operaciones numéricas. Consulta los CÓDIGOS COMPLETOS aquí.

si no os.environ.get(“OPENAI_API_KEY”): os.environ[“OPENAI_API_KEY”] = getpass.getpass(“Ingrese OPENAI_API_KEY: “) MODELO = os.environ.get(“OPENAI_MODEL”, “gpt-4o-mini”) EMB_MODEL = os.environ.get(“OPENAI_EMBED_MODEL”, “text-embedding-3-small”) llm_fast = ChatOpenAI(modelo=MODEL, temperatura=0) llm_deep = ChatOpenAI(modelo=MODELO, temperatura=0) llm_reflect = ChatOpenAI(modelo=MODELO, temperatura=0) emb = OpenAIEmbeddings(modelo=EMB_MODEL)

Cargamos de forma segura la clave API de OpenAI en tiempo de ejecución e inicializamos los modelos de lenguaje utilizados para un razonamiento rápido, profundo y reflexivo. También configuramos el modelo de incrustación que impulsa la similitud semántica en la memoria. Esta separación nos permite cambiar de manera flexible la profundidad del razonamiento mientras mantenemos un espacio de representación compartido para la memoria. Consulta los CÓDIGOS COMPLETOS aquí.

clase Nota (BaseModel): note_id: título de cadena: contenido de cadena: etiquetas de cadena: Lista[str] = Campo (default_factory = lista) creado_en_unix: contexto flotante: Dict[str, Any] = Campo(default_factory=dict) clase MemoryGraph: def __init__(self): self.g = nx.Graph() self.note_vectors = {} def _cos(self, a, b): return float(np.dot(a, b) / ((np.linalg.norm(a) + 1e-9) * (np.linalg.norm(b) + 1e-9))) def add_note(self, note, vec): self.g.add_node(note.note_id, **note.model_dump()) self.note_vectors[note.note_id] = vec def topk_relacionado(self, vec, k=5): puntuado = [(nid, self._cos(vec, v)) for nid, v in self.note_vectors.items()]
anotado.sort(clave=lambda x: x[1]reverso=Verdadero) retorno [{“note_id”: n, “score”: s, “title”: self.g.nodes[n][“title”]} para n, s en puntuación[:k]]def link_note(self, a, b, w, r): si a != b: self.g.add_edge(a, b, peso=w, razón=r) def evolucionar_links(self, nid, vec): para r en self.topk_relacionado(vec, 8): si r[“score”] >= 0,78: self.link_note(nid,r[“note_id”]r.[“score”]”evolucionar”) MEM = MemoryGraph()

Construimos un gráfico de memoria agente inspirado en el método Zettelkasten, donde cada interacción se almacena como una nota atómica. Incorporamos cada nota y la conectamos con notas relacionadas semánticamente mediante puntuaciones de similitud. Consulta los CÓDIGOS COMPLETOS aquí.

@tool def web_get(url: str) -> str: importar urllib.request con urllib.request.urlopen(url, timeout=15) como r: devolver r.read(25000).decode(“utf-8″, errores=”ignore”) @tool def Memory_search(query: str, k: int = 5) -> str: qv = np.array(emb.embed_query(query)) hits = MEM.topk_ related(qv, k) return json.dumps(hits, asegurar_ascii=False) @tool def Memory_neighbors(note_id: str) -> str: si note_id no está en MEM.g: return “[]” devolver json.dumps ([
{“note_id”: n, “weight”: MEM.g[note_id][n][“weight”]} para n en MEM.g.neighbors(note_id) ]) HERRAMIENTAS = [web_get, memory_search, memory_neighbors]
TOOLS_BY_NAME = {t.name: t para t en HERRAMIENTAS}

Definimos las herramientas externas que el agente puede invocar, incluido el acceso web y la recuperación basada en memoria. Integramos estas herramientas de forma estructurada para que el agente pueda consultar experiencias pasadas o obtener nueva información cuando sea necesario. Consulta los CÓDIGOS COMPLETOS aquí.

clase DeliberationDecision(BaseModel): modo: Literal[“fast”, “deep”]
motivo: str pasos_sugeridos: Lista[str]

clase RunSpec(BaseModel): objetivo: restricciones str: Lista[str]
deliveryable_format: str must_use_memory: bool max_tool_calls: int clase Reflexión (BaseModel): note_title: str note_tags: Lista[str]
nuevas_reglas: lista[str]
what_worked: Lista[str]
what_failed: Lista[str]

clase AgentState (TypedDict, total = Falso): run_spec: Dict[str, Any]
mensajes: anotados[List[AnyMessage]operador.add]decisión: Dict[str, Any]
final: str Budget_calls_remaining: int tool_calls_used: int max_tool_calls: int last_note_id: str DECIDER_SYS = “Decide rápido o profundo”. AGENT_FAST = “Operar rápido.” AGENT_DEEP = “Operar en profundidad.” REFLECT_SYS = “Reflexionar y almacenar aprendizajes.”

Formalizamos las representaciones internas del agente utilizando esquemas estructurados para la deliberación, los objetivos de ejecución, la reflexión y el estado global. También definimos las indicaciones del sistema que guían el comportamiento en modos rápido y profundo. Esto garantiza que el razonamiento y las decisiones del agente sigan siendo coherentes, interpretables y controlables. Consulta los CÓDIGOS COMPLETOS aquí.

def deliberado(st): especificación = RunSpec.model_validate(st[“run_spec”]) d = llm_fast.with_structured_output(DeliberaciónDecisión).invocar([
SystemMessage(content=DECIDER_SYS),
HumanMessage(content=json.dumps(spec.model_dump()))
]) devolver {“decisión”: d.model_dump(), “budget_calls_remaining”: st[“budget_calls_remaining”] – 1} def agente(st): especificación = RunSpec.model_validate(st[“run_spec”]) d = DecisiónDeliberación.model_validate(st[“decision”]) llm = llm_deep if d.mode == “deep” else llm_fast sys = AGENT_DEEP if d.mode == “deep” else AGENT_FAST out = llm.bind_tools(TOOLS).invoke([
SystemMessage(content=sys),
*st.get(“messages”, []), HumanMessage(content=json.dumps(spec.model_dump())) ]) return {“mensajes”: [out]”llamadas_presupuestarias_restantes”: st[“budget_calls_remaining”] – 1} def ruta(st): devuelve “herramientas” si st[“messages”][-1].tool_calls else “finalizar” def tools_node(st): msgs = []
usado = st.get(“tool_calls_used”, 0) para c en st[“messages”][-1].tool_calls: obs = HERRAMIENTAS_POR_NOMBRE[c[“name”]].invocar(c[“args”]) msgs.append(ToolMessage(content=str(obs), tool_call_id=c[“id”])) usado += 1 return {“mensajes”: mensajes, “tool_calls_used”: usado} def finalizar(st): out = llm_deep.invoke(st[“messages”] + [HumanMessage(content=”Return final output”)]) return {“final”: out.content} def reflejar(st): r = llm_reflect.with_structured_output(Reflexión).invoke([
SystemMessage(content=REFLECT_SYS),
HumanMessage(content=st[“final”]) ]) nota = Nota( note_id=str(time.time()), título=r.note_title, contenido=st[“final”]etiquetas=r.note_tags, create_at_unix=time.time() ) vec = np.array(emb.embed_query(note.title + note.content)) MEM.add_note(nota, vec) MEM.evolve_links(note.note_id, vec) return {“last_note_id”: note.note_id}

Implementamos los comportamientos agentes centrales como nodos LangGraph, incluida la deliberación, la acción, la ejecución de herramientas, la finalización y la reflexión. Orquestamos cómo fluye la información entre estas etapas y cómo las decisiones afectan la ruta de ejecución. Consulta los CÓDIGOS COMPLETOS aquí.

g = StateGraph(AgentState) g.add_node(“deliberar”, deliberar) g.add_node(“agente”, agente) g.add_node(“herramientas”, herramientas_node) g.add_node(“finalizar”, finalizar) g.add_node(“reflexionar”, reflejar) g.add_edge(START, “deliberar”) g.add_edge(“deliberar”, “agente”) g.add_conditional_edges(“agente”, ruta, [“tools”, “finalize”]) g.add_edge(“tools”, “agent”) g.add_edge(“finalize”, “reflect”) g.add_edge(“reflect”, END) graph = g.compile(checkpointer=InMemorySaver()) def run_agent(goal, constraints=Ninguno, thread_id=”demo”): si las restricciones son Ninguna: restricciones = []
especificación = RunSpec( objetivo=meta, restricciones=restricciones, delivery_format=”markdown”, must_use_memory=True, max_tool_calls=6 ).model_dump() return graph.invoke({ “run_spec”: especificación, “mensajes”: []”budget_calls_remaining”: 10, “tool_calls_used”: 0, “max_tool_calls”: 6 }, config={“configurable”: {“thread_id”: thread_id}})

Montamos todos los nodos en un flujo de trabajo LangGraph y lo compilamos con gestión de estado controlada. También definimos una función de ejecución reutilizable que ejecuta el agente mientras preserva la memoria entre ejecuciones.

En conclusión, mostramos cómo un agente puede mejorar continuamente su comportamiento a través de la reflexión y la memoria en lugar de depender de indicaciones estáticas o lógica codificada. Usamos LangGraph para orquestar la deliberación, la ejecución, el gobierno de herramientas y la reflexión como un gráfico coherente, mientras que los modelos OpenAI brindan capacidades de razonamiento y síntesis en cada etapa. Este enfoque ilustró cómo los sistemas de IA agentes pueden acercarse a la autonomía adaptando su profundidad de razonamiento, reutilizando conocimientos previos y codificando lecciones como memoria persistente, formando una base práctica para construir agentes escalables y de mejora personal en aplicaciones del mundo real.

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.

Consulte nuestra última versión de ai2025.dev, una plataforma de análisis centrada en 2025 que convierte los lanzamientos de modelos, los puntos de referencia y la actividad del ecosistema en un conjunto de datos estructurado que puede filtrar, comparar y exportar.

Asif Razzaq es el director ejecutivo de Marktechpost Media Inc.. Como empresario e ingeniero visionario, Asif está comprometido a aprovechar el potencial de la inteligencia artificial para el bien social. Su esfuerzo más reciente es el lanzamiento de una plataforma de medios de inteligencia artificial, Marktechpost, que se destaca por su cobertura en profundidad del aprendizaje automático y las noticias sobre aprendizaje profundo que es técnicamente sólida y fácilmente comprensible para una amplia audiencia. La plataforma cuenta con más de 2 millones de visitas mensuales, lo que ilustra su popularidad entre el público.