<RETURN_TO_BASE

Designing an Advanced Agentic AI Architecture

Learn to build an advanced Agentic AI system with LangGraph and OpenAI.

Overview

In this tutorial, we build a genuinely advanced Agentic AI system using LangGraph and OpenAI models by going beyond simple planner-executor loops. We implement adaptive deliberation, where the agent dynamically decides between fast and deep reasoning; a Zettelkasten-style agentic memory graph that stores atomic knowledge and automatically links related experiences; and a governed tool-use mechanism that enforces constraints during execution.

Key Components

By combining structured state management, memory-aware retrieval, reflexive learning, and controlled tool invocation, we demonstrate how modern agentic systems can reason, act, learn, and evolve rather than respond in a single pass.

Setting Up the Environment

To set up the execution environment, we install all required libraries and import the core modules. We integrate LangGraph for orchestration, LangChain for model and tool abstractions, and supporting libraries for memory graphs and numerical operations.

!pip -q install -U langgraph langchain-openai langchain-core pydantic numpy networkx requests
 
import os, getpass, json, time, operator
from typing import List, Dict, Any, Optional, Literal
from typing_extensions import TypedDict, Annotated
import numpy as np
import networkx as nx
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.messages import SystemMessage, HumanMessage, ToolMessage, AnyMessage
from langchain_core.tools import tool
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver

Loading OpenAI API Key

We securely load the OpenAI API key at runtime and initialize the language models used for fast, deep, and reflective reasoning. We also configure the embedding model that powers semantic similarity in memory.

if not os.environ.get("OPENAI_API_KEY"):
   os.environ["OPENAI_API_KEY"] = getpass.getpass("Enter OPENAI_API_KEY: ")
 
MODEL = os.environ.get("OPENAI_MODEL", "gpt-4o-mini")
EMB_MODEL = os.environ.get("OPENAI_EMBED_MODEL", "text-embedding-3-small")
 
llm_fast = ChatOpenAI(model=MODEL, temperature=0)
llm_deep = ChatOpenAI(model=MODEL, temperature=0)
llm_reflect = ChatOpenAI(model=MODEL, temperature=0)
emb = OpenAIEmbeddings(model=EMB_MODEL)

Creating the Agentic Memory Graph

We construct an agentic memory graph inspired by the Zettelkasten method, where each interaction is stored as an atomic note.

class Note(BaseModel):
   note_id: str
   title: str
   content: str
   tags: List[str] = Field(default_factory=list)
   created_at_unix: float
   context: Dict[str, Any] = Field(default_factory=dict)
 
class 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_related(self, vec, k=5):
       scored = [(nid, self._cos(vec, v)) for nid, v in self.note_vectors.items()]
       scored.sort(key=lambda x: x[1], reverse=True)
       return [{"note_id": n, "score": s, "title": self.g.nodes[n]["title"]} for n, s in scored[:k]]
 
   def link_note(self, a, b, w, r):
       if a != b:
           self.g.add_edge(a, b, weight=w, reason=r)
 
   def evolve_links(self, nid, vec):
       for r in self.topk_related(vec, 8):
           if r["score"] >= 0.78:
               self.link_note(nid, r["note_id"], r["score"], "evolve")
MEM = MemoryGraph()

Implementing External Tools

We define the external tools the agent can invoke, such as web access and memory-based retrieval.

@tool
def web_get(url: str) -> str:
   import urllib.request
   with urllib.request.urlopen(url, timeout=15) as r:
       return r.read(25000).decode("utf-8", errors="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, ensure_ascii=False)
 
@tool
def memory_neighbors(note_id: str) -> str:
   if note_id not in MEM.g:
       return "[]"
   return json.dumps([
       {"note_id": n, "weight": MEM.g[note_id][n]["weight"]}
       for n in MEM.g.neighbors(note_id)
   ])
TOOLS = [web_get, memory_search, memory_neighbors]
TOOLS_BY_NAME = {t.name: t for t in TOOLS}

Core Agentic Behaviors

The core agentic behaviors, including deliberation, action, tool execution, finalization, and reflection, are implemented as LangGraph nodes.

def deliberate(st):
   spec = RunSpec.model_validate(st["run_spec"])
   d = llm_fast.with_structured_output(DeliberationDecision).invoke([
       SystemMessage(content=DECIDER_SYS),
       HumanMessage(content=json.dumps(spec.model_dump()))
   ])
   return {"decision": d.model_dump(), "budget_calls_remaining": st["budget_calls_remaining"] - 1}

Conclusion

We illustrated how agentic AI systems can move closer to autonomy by adapting their reasoning depth and reusing prior knowledge. This approach forms a practical foundation for building scalable, self-improving agents in real-world applications.

For full implementation, check out the FULL CODES here.

🇷🇺

Сменить язык

Читать эту статью на русском

Переключить на Русский