Methods to Design an Agentic AI Structure with LangGraph and OpenAI Utilizing Adaptive Deliberation, Reminiscence Graphs, and Reflexion Loops

On this tutorial, we construct a genuinely superior Agentic AI system utilizing LangGraph and OpenAI fashions by going past easy planner, executor loops. We implement adaptive deliberation, the place the agent dynamically decides between quick and deep reasoning; a Zettelkasten-style agentic reminiscence graph that shops atomic information and mechanically hyperlinks associated experiences; and a ruled tool-use mechanism that enforces constraints throughout execution. By combining structured state administration, memory-aware retrieval, reflexive studying, and managed instrument invocation, we show how trendy agentic methods can motive, act, study, and evolve somewhat than reply in a single cross. Take a look at the FULL CODES right here.

!pip -q set up -U langgraph langchain-openai langchain-core pydantic numpy networkx requests


import os, getpass, json, time, operator
from typing import Record, Dict, Any, Optionally available, Literal
from typing_extensions import TypedDict, Annotated
import numpy as np
import networkx as nx
from pydantic import BaseModel, Discipline
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.messages import SystemMessage, HumanMessage, ToolMessage, AnyMessage
from langchain_core.instruments import instrument
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.reminiscence import InMemorySaver

We arrange the execution surroundings by putting in all required libraries and importing the core modules. We carry collectively LangGraph for orchestration, LangChain for mannequin and gear abstractions, and supporting libraries for reminiscence graphs and numerical operations. Take a look at the FULL CODES right here.

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(mannequin=MODEL, temperature=0)
llm_deep = ChatOpenAI(mannequin=MODEL, temperature=0)
llm_reflect = ChatOpenAI(mannequin=MODEL, temperature=0)
emb = OpenAIEmbeddings(mannequin=EMB_MODEL)

We securely load the OpenAI API key at runtime and initialize the language fashions used for quick, deep, and reflective reasoning. We additionally configure the embedding mannequin that powers semantic similarity in reminiscence. This separation permits us to flexibly change reasoning depth whereas sustaining a shared illustration area for reminiscence. Take a look at the FULL CODES right here.

class Notice(BaseModel):
   note_id: str
   title: str
   content material: str
   tags: Record[str] = Discipline(default_factory=record)
   created_at_unix: float
   context: Dict[str, Any] = Discipline(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, be aware, vec):
       self.g.add_node(be aware.note_id, **be aware.model_dump())
       self.note_vectors[note.note_id] = vec


   def topk_related(self, vec, okay=5):
       scored = [(nid, self._cos(vec, v)) for nid, v in self.note_vectors.items()]
       scored.kind(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, motive=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()

We assemble an agentic reminiscence graph impressed by the Zettelkasten technique, the place every interplay is saved as an atomic be aware. We embed every be aware and join it to semantically associated notes utilizing similarity scores. Take a look at the FULL CODES right here.

@instrument
def web_get(url: str) -> str:
   import urllib.request
   with urllib.request.urlopen(url, timeout=15) as r:
       return r.learn(25000).decode("utf-8", errors="ignore")


@instrument
def memory_search(question: str, okay: int = 5) -> str:
   qv = np.array(emb.embed_query(question))
   hits = MEM.topk_related(qv, okay)
   return json.dumps(hits, ensure_ascii=False)


@instrument
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.identify: t for t in TOOLS}

We outline the exterior instruments the agent can invoke, together with net entry and memory-based retrieval. We combine these instruments in a structured approach so the agent can question previous experiences or fetch new info when obligatory. Take a look at the FULL CODES right here.

class DeliberationDecision(BaseModel):
   mode: Literal["fast", "deep"]
   motive: str
   suggested_steps: Record[str]


class RunSpec(BaseModel):
   purpose: str
   constraints: Record[str]
   deliverable_format: str
   must_use_memory: bool
   max_tool_calls: int


class Reflection(BaseModel):
   note_title: str
   note_tags: Record[str]
   new_rules: Record[str]
   what_worked: Record[str]
   what_failed: Record[str]


class AgentState(TypedDict, complete=False):
   run_spec: Dict[str, Any]
   messages: Annotated[List[AnyMessage], operator.add]
   resolution: Dict[str, Any]
   remaining: str
   budget_calls_remaining: int
   tool_calls_used: int
   max_tool_calls: int
   last_note_id: str


DECIDER_SYS = "Determine quick vs deep."
AGENT_FAST = "Function quick."
AGENT_DEEP = "Function deep."
REFLECT_SYS = "Mirror and retailer learnings."

We formalize the agent’s inner representations utilizing structured schemas for deliberation, execution targets, reflection, and world state. We additionally outline the system prompts that information conduct in quick and deep modes. This ensures the agent’s reasoning and choices stay constant, interpretable, and controllable. Take a look at the FULL CODES right here.

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 {"resolution": d.model_dump(), "budget_calls_remaining": st["budget_calls_remaining"] - 1}


def agent(st):
   spec = RunSpec.model_validate(st["run_spec"])
   d = DeliberationDecision.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 material=json.dumps(spec.model_dump()))
   ])
   return {"messages": [out], "budget_calls_remaining": st["budget_calls_remaining"] - 1}


def route(st):
   return "instruments" if st["messages"][-1].tool_calls else "finalize"


def tools_node(st):
   msgs = []
   used = st.get("tool_calls_used", 0)
   for c in st["messages"][-1].tool_calls:
       obs = TOOLS_BY_NAME[c["name"]].invoke(c["args"])
       msgs.append(ToolMessage(content material=str(obs), tool_call_id=c["id"]))
       used += 1
   return {"messages": msgs, "tool_calls_used": used}


def finalize(st):
   out = llm_deep.invoke(st["messages"] + [HumanMessage(content="Return final output")])
   return {"remaining": out.content material}


def replicate(st):
   r = llm_reflect.with_structured_output(Reflection).invoke([
       SystemMessage(content=REFLECT_SYS),
       HumanMessage(content=st["final"])
   ])
   be aware = Notice(
       note_id=str(time.time()),
       title=r.note_title,
       content material=st["final"],
       tags=r.note_tags,
       created_at_unix=time.time()
   )
   vec = np.array(emb.embed_query(be aware.title + be aware.content material))
   MEM.add_note(be aware, vec)
   MEM.evolve_links(be aware.note_id, vec)
   return {"last_note_id": be aware.note_id}

We implement the core agentic behaviors as LangGraph nodes, together with deliberation, motion, instrument execution, finalization, and reflection. We orchestrate how info flows between these levels and the way choices have an effect on the execution path. Take a look at the FULL CODES right here.

g = StateGraph(AgentState)
g.add_node("deliberate", deliberate)
g.add_node("agent", agent)
g.add_node("instruments", tools_node)
g.add_node("finalize", finalize)
g.add_node("replicate", replicate)


g.add_edge(START, "deliberate")
g.add_edge("deliberate", "agent")
g.add_conditional_edges("agent", route, ["tools", "finalize"])
g.add_edge("instruments", "agent")
g.add_edge("finalize", "replicate")
g.add_edge("replicate", END)


graph = g.compile(checkpointer=InMemorySaver())


def run_agent(purpose, constraints=None, thread_id="demo"):
   if constraints is None:
       constraints = []
   spec = RunSpec(
       purpose=purpose,
       constraints=constraints,
       deliverable_format="markdown",
       must_use_memory=True,
       max_tool_calls=6
   ).model_dump()


   return graph.invoke({
       "run_spec": spec,
       "messages": [],
       "budget_calls_remaining": 10,
       "tool_calls_used": 0,
       "max_tool_calls": 6
   }, config={"configurable": {"thread_id": thread_id}})

We assemble all nodes right into a LangGraph workflow and compile it with checkpointed state administration. We additionally outline a reusable runner perform that executes the agent whereas preserving reminiscence throughout runs.

In conclusion, we confirmed how an agent can constantly enhance its conduct by way of reflection and reminiscence somewhat than counting on static prompts or hard-coded logic. We used LangGraph to orchestrate deliberation, execution, instrument governance, and reflexion as a coherent graph, whereas OpenAI fashions present the reasoning and synthesis capabilities at every stage. This strategy illustrated how agentic AI methods can transfer nearer to autonomy by adapting their reasoning depth, reusing prior information, and encoding classes as persistent reminiscence, forming a sensible basis for constructing scalable, self-improving brokers in real-world functions.

Take a look at the FULL CODES right here. Additionally, be happy to comply with us on Twitter and don’t overlook to hitch our 100k+ ML SubReddit and Subscribe to our Publication. Wait! are you on telegram? now you’ll be able to be part of us on telegram as effectively.

Take a look at our newest launch of ai2025.dev, a 2025-focused analytics platform that turns mannequin launches, benchmarks, and ecosystem exercise right into a structured dataset you’ll be able to filter, examine, and export

Asif Razzaq is the CEO of Marktechpost Media Inc.. As a visionary entrepreneur and engineer, Asif is dedicated to harnessing the potential of Synthetic Intelligence for social good. His most up-to-date endeavor is the launch of an Synthetic Intelligence Media Platform, Marktechpost, which stands out for its in-depth protection of machine studying and deep studying information that’s each technically sound and simply comprehensible by a large viewers. The platform boasts of over 2 million month-to-month views, illustrating its recognition amongst audiences.