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 InMemorySaverLoading 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.
Сменить язык
Читать эту статью на русском