Foreword: A Quiet Revolution with Loud Consequences
There is a moment in every technological revolution when the change stops being theoretical and starts being personal. You feel it in the way a task that used to take you three hours now takes twelve minutes. You feel it when a system remembers something you told it six weeks ago and applies that knowledge without being asked. You feel it when you realize the "assistant" you are working with is not one mind but a whole coordinated team of minds — each one a specialist, each one doing its job with a kind of quiet, tireless precision that no human team could sustain around the clock.
That moment is happening right now, in August 2026, and it is happening faster than most people expected.
This article is an attempt to slow that moment down and look at it carefully. We will walk through the major architectural shifts that are turning large language models from clever text generators into genuine cognitive partners. We will look at how specialized teams of AI agents are being assembled to tackle complex, multi-step workflows. We will explore the engineering of long-term memory, the staggering expansion of context windows, and the practical code that makes all of it real. Along the way, we will keep one foot firmly planted in the concrete world of working software, because the best way to understand these ideas is to see them in action.
Chapter 1: From Chatbot to Cognitive Partner — The Evolution of the LLM
To appreciate where we are, it helps to remember where we started. When the first generation of large language models arrived in the public consciousness around 2020 and 2021, they were astonishing and limited in equal measure. They could write poetry, summarize documents, answer trivia questions, and generate code that sometimes worked. But they were, at their core, stateless text completion engines. You gave them a prompt, they gave you a response, and then they forgot everything. Every conversation started from zero. Every session was an island.
The metaphor that kept coming up in those early days was the goldfish. The model had no memory beyond the current conversation window, and even that window was small — typically a few thousand tokens, which is roughly equivalent to a few pages of text. If you were working on a complex project and needed the model to remember something from last Tuesday, you were out of luck. You had to paste the relevant context back in yourself, every single time.
That limitation shaped everything. It shaped the kinds of tasks people used these models for, the kinds of products companies built on top of them, and the kinds of expectations people brought to the interaction. LLMs were tools, not partners. Powerful tools, certainly, but tools in the same category as a very sophisticated search engine or a very fast calculator.
What changed? Several things changed simultaneously, and their convergence is what makes the current moment so interesting.
The most architecturally significant shift was the move from passive text generation to active agency. Researchers and engineers began to think seriously about what it would mean to give an LLM the ability to take actions in the world — not just generate text about actions. This led to the concept of the "agent": a model that can call external tools, browse the web, write and execute code, query databases, send emails, and generally interact with the world beyond the conversation window. The model becomes the brain of a system, and the system has hands.
Running in parallel was the context window explosion. In 2023, a context window of 32,000 tokens felt generous. By early 2025, Google's Gemini 2.5 Pro offered one million tokens. Meta's Llama 4 Scout reached ten million tokens in April 2025. By August 2026, leading frontier models routinely operate with context windows exceeding twenty million tokens, and the practical ceiling continues to rise. These are not incremental improvements — a ten-million-token context window can hold entire codebases, entire legal contracts, entire research libraries, all at once, all available to the model in a single pass.
Memory was the third piece of the puzzle. Engineers began building systems that could store information outside the model's context window and retrieve it on demand, giving agents a form of long-term memory that persists across sessions. OpenAI's April 2025 update introduced a dual-memory system that reduced repeated user prompts by 78 percent. Google released its Memory Bank system in July 2025. Academic researchers presented the A-MEM system at NeurIPS 2025, which dynamically organizes memories without relying on predefined structures, drawing inspiration from the Zettelkasten method of knowledge management. By 2026, persistent agent memory has become a baseline expectation rather than a premium feature.
What tied all of this together was specialization. Instead of asking one general-purpose model to do everything, engineers began assembling teams of specialized agents — each one optimized for a particular kind of task — and orchestrating them to collaborate on complex goals. This is the multi-agent paradigm, and it is arguably the most transformative shift of all.
Let us look at each of these changes in depth, starting with the one that makes everything else possible: the agent itself.
Chapter 2: The Anatomy of an AI Agent — What Makes a Model into an Actor
The word "agent" has a specific technical meaning in this context, and it is worth being precise about it. An AI agent is a system in which a large language model serves as the reasoning core, but the system as a whole can perceive its environment, make decisions, take actions, and observe the results of those actions in a loop. This loop — often called the "ReAct" loop (from Reasoning and Acting) — is what distinguishes an agent from a simple chatbot.
The loop works like this. The agent receives a goal or a query. It reasons about what steps are needed to achieve that goal. It selects a tool or action to execute. It executes the action and observes the result. It reasons again about what to do next, given the new information. It continues until the goal is achieved or until it determines that the goal cannot be achieved.
This sounds simple, but the implications are profound. An agent can browse the web to find current information. It can write a Python script, execute it, read the output, fix any errors, and execute it again. It can query a database, analyze the results, and generate a report. It can send an email, wait for a reply, and adjust its behavior based on the reply. It can coordinate with other agents, delegating subtasks and aggregating results.
The key components of any agent system are the model itself, the tools it has access to, the memory systems that give it continuity, and the orchestration layer that manages its behavior. The most fundamental of these — the one everything else depends on — is tool calling.
Tool calling is the mechanism by which a language model signals that it wants to invoke an external function. The model does not execute the function itself. Instead, it generates a structured description of the function call it wants to make, including the function name and the arguments it wants to pass. The surrounding application code intercepts this description, executes the actual function, and feeds the result back to the model as part of the conversation.
The following example demonstrates a complete tool-calling loop using the OpenAI API with GPT-5.6 Sol, the current flagship model as of August 2026. Notice how the model acts as a decision-maker while the Python code acts as the executor. This separation of concerns is fundamental to the entire agentic paradigm.
Install dependencies:
pip install openai>=3.0.0
Set your API key:
export OPENAI_API_KEY="your-key-here"
Run:
python agent_tool_calling.py
"""
agent_tool_calling.py
Demonstrates a complete agentic tool-calling loop using the OpenAI API.
The model decides WHEN to call a tool; Python decides HOW to execute it.
This pattern is the foundation of every agentic system.
Tested with: openai>=3.0.0, Python 3.10+
"""
import json
import os
from openai import OpenAI
# ---------------------------------------------------------------------------
# Client initialization. The API key is read from the environment, which is
# the correct approach for any production or shared codebase. Never hard-code
# credentials in source files.
# ---------------------------------------------------------------------------
client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))
# ---------------------------------------------------------------------------
# Tool implementations. These are ordinary Python functions. The model never
# sees this code directly; it only sees the JSON schema descriptions below.
# ---------------------------------------------------------------------------
def get_project_status(project_id: str) -> dict:
"""
Simulates fetching the status of an internal project from a database.
In a real system, this would query an actual project management API.
Args:
project_id: The unique identifier for the project.
Returns:
A dictionary containing project metadata and current status.
"""
projects = {
"PRJ-001": {
"name": "Industrial Digital Twin Platform",
"status": "On Track",
"completion_percentage": 67,
"next_milestone": "Integration testing",
"due_date": "2026-10-15",
},
"PRJ-002": {
"name": "Smart Grid Analytics Module",
"status": "At Risk",
"completion_percentage": 42,
"next_milestone": "Data pipeline validation",
"due_date": "2026-09-01",
},
}
return projects.get(
project_id,
{"error": f"Project {project_id} not found in the system."},
)
def calculate_resource_allocation(team_size: int, weeks: int) -> dict:
"""
Calculates resource allocation metrics for project planning.
Args:
team_size: Number of engineers on the team.
weeks: Duration of the planning period in weeks.
Returns:
A dictionary with allocation metrics.
"""
total_person_weeks = team_size * weeks
# Assuming 80% effective utilization after meetings and overhead
effective_person_weeks = round(total_person_weeks * 0.80, 1)
return {
"team_size": team_size,
"planning_weeks": weeks,
"total_person_weeks": total_person_weeks,
"effective_person_weeks": effective_person_weeks,
"utilization_rate": "80%",
}
# ---------------------------------------------------------------------------
# Tool schema definitions. This is what the model actually reads. The quality
# of these descriptions directly determines how accurately the model will
# decide when and how to call each tool. Write them like good documentation.
# ---------------------------------------------------------------------------
TOOLS = [
{
"type": "function",
"function": {
"name": "get_project_status",
"description": (
"Retrieves the current status, completion percentage, next "
"milestone, and due date for a given internal project. Use "
"this when the user asks about the progress or health of a "
"specific project."
),
"parameters": {
"type": "object",
"properties": {
"project_id": {
"type": "string",
"description": (
"The unique project identifier, formatted as "
"PRJ-XXX, e.g. PRJ-001."
),
}
},
"required": ["project_id"],
},
},
},
{
"type": "function",
"function": {
"name": "calculate_resource_allocation",
"description": (
"Calculates resource allocation and effective person-weeks "
"for a project team over a given number of weeks. Use this "
"when the user asks about capacity planning or team bandwidth."
),
"parameters": {
"type": "object",
"properties": {
"team_size": {
"type": "integer",
"description": "The number of engineers on the team.",
},
"weeks": {
"type": "integer",
"description": "The number of weeks in the planning period.",
},
},
"required": ["team_size", "weeks"],
},
},
},
]
# Registry maps function names to their Python implementations
FUNCTION_REGISTRY = {
"get_project_status": get_project_status,
"calculate_resource_allocation": calculate_resource_allocation,
}
def run_agent(user_query: str) -> str:
"""
Runs a single-turn agentic loop: the model reasons, calls tools if
needed, and produces a final natural-language response.
Args:
user_query: The user's question or request.
Returns:
The model's final natural-language response as a string.
"""
messages = [
{
"role": "system",
"content": (
"You are a helpful project management assistant. You have "
"access to tools that can retrieve project status and "
"calculate resource allocation. Always use the tools when "
"the user asks for specific project or resource data."
),
},
{"role": "user", "content": user_query},
]
# First model call: let the model decide whether to use a tool
response = client.chat.completions.create(
model="gpt-5", # Current flagship model, August 2025+
messages=messages,
tools=TOOLS,
tool_choice="auto", # The model decides autonomously
)
response_message = response.choices[0].message
messages.append(response_message)
# Handle tool calls if the model requested them
if response_message.tool_calls:
for tool_call in response_message.tool_calls:
function_name = tool_call.function.name
function_args = json.loads(tool_call.function.arguments)
# Execute the actual Python function
function_result = FUNCTION_REGISTRY[function_name](**function_args)
# Feed the result back into the conversation
messages.append(
{
"role": "tool",
"tool_call_id": tool_call.id,
"name": function_name,
"content": json.dumps(function_result),
}
)
# Second model call: synthesize tool results into a response
final_response = client.chat.completions.create(
model="gpt-5",
messages=messages,
)
return final_response.choices[0].message.content
# If no tool was called, return the direct response
return response_message.content
if __name__ == "__main__":
query = (
"What is the status of project PRJ-002, and if I have a team of "
"8 engineers for 12 weeks, how much effective capacity do I have?"
)
print(f"User: {query}\n")
print(f"Agent: {run_agent(query)}")
The code above is doing something deceptively elegant. The model reads the tool schemas and the user's question, and it figures out on its own that it needs to call both tools to answer the question fully. It generates two tool calls in a single response, the application executes both Python functions, feeds both results back, and the model weaves them into a coherent, natural-language answer. The user never sees any of the JSON. They just get an answer that feels like it came from a knowledgeable colleague who had access to the right systems.
This is the foundational pattern. Everything else in agentic AI is built on top of this loop, made more sophisticated, more persistent, and more parallel.
Chapter 3: Running Intelligence Locally — The Ollama Approach
Not every organization wants to send its data to a cloud provider. Data privacy regulations, corporate security policies, latency requirements, and simple cost considerations all push toward running models locally. This is where tools like Ollama have become genuinely transformative.
Ollama is an open-source platform that lets you run large language models on your own hardware, from a developer laptop to an on-premises server cluster. It provides a clean, OpenAI-compatible REST API, which means that code written for the OpenAI API can often be pointed at a local Ollama instance with minimal changes. Models like Llama 4 Scout, Phi-4-mini, Gemma 4, and Qwen can be pulled and run locally with a single command.
The practical workflow is straightforward. You install Ollama, pull a model, and start the server. From that point on, your Python code talks to a local HTTP endpoint instead of a remote cloud service. No data leaves your machine. No API costs accumulate. No rate limits apply.
Install Ollama:
# macOS / Linux
curl -fsSL https://ollama.com/install.sh | sh
# Pull the models used in this chapter
ollama pull llama4:scout
ollama pull phi4-mini
# Start the server (runs on http://localhost:11434 by default)
ollama serve
Install Python dependencies:
pip install ollama>=0.32.0
Run:
python local_agent_ollama.py
The following example demonstrates a complete local agent setup using Ollama. It mirrors the structure of the previous example but runs entirely on local infrastructure. Notice how the abstraction layer makes the switch almost invisible at the code level.
"""
local_agent_ollama.py
A complete agentic tool-calling system running on a LOCAL LLM via Ollama.
No data leaves the machine. No API key required.
Tested with: ollama>=0.32.0, Python 3.10+
Models required: ollama pull llama4:scout
"""
import json
import ollama
# ---------------------------------------------------------------------------
# Tool implementations. These are identical to what you would write for a
# cloud-based agent. The local/remote distinction lives only in the client
# initialization, not in the business logic.
# ---------------------------------------------------------------------------
def search_knowledge_base(query: str, domain: str = "general") -> dict:
"""
Simulates searching an internal knowledge base for relevant documents.
In production, this would query a vector database like ChromaDB or Qdrant.
Args:
query: The search query string.
domain: The knowledge domain to search (e.g., 'engineering', 'hr').
Returns:
A dictionary containing matching document excerpts and metadata.
"""
knowledge_base = {
"engineering": {
"query_match": query,
"documents": [
{
"title": "IEC PLC Programming Standards v3.2",
"excerpt": (
"All automation engineering projects must follow IEC 61131-3 "
"structured text conventions. Function blocks must "
"include inline documentation and unit tests."
),
"relevance_score": 0.94,
},
{
"title": "Industrial IoT Integration Guidelines",
"excerpt": (
"OPC-UA is the preferred protocol for machine-to-cloud "
"communication. MQTT may be used for edge-to-edge "
"messaging where latency is critical."
),
"relevance_score": 0.87,
},
],
},
"general": {
"query_match": query,
"documents": [
{
"title": "General Policy Document",
"excerpt": "No specific results found for this query.",
"relevance_score": 0.40,
}
],
},
}
return knowledge_base.get(domain, knowledge_base["general"])
def summarize_meeting_notes(notes: str, output_format: str = "bullet") -> dict:
"""
Processes raw meeting notes and extracts key action items.
In a real system, this might call a specialized summarization pipeline.
Args:
notes: The raw meeting notes text.
output_format: Either 'bullet' for bullet points or 'prose' for
a paragraph summary.
Returns:
A dictionary containing the summary and extracted action items.
"""
word_count = len(notes.split())
return {
"original_word_count": word_count,
"output_format": output_format,
"summary": (
f"Meeting covered {word_count} words of discussion. "
"Key decisions were made regarding project timelines and "
"resource allocation. Follow-up required from engineering team."
),
"action_items": [
"Engineering team to deliver prototype by end of sprint.",
"PM to update stakeholder dashboard by Friday.",
"Legal review of vendor contract to be completed within 5 days.",
],
}
# ---------------------------------------------------------------------------
# Tool schema definitions in the format that Ollama's tool-calling interface
# expects. This format is compatible with the OpenAI function-calling spec,
# which is why switching between local and remote models is so seamless.
# ---------------------------------------------------------------------------
TOOLS = [
{
"type": "function",
"function": {
"name": "search_knowledge_base",
"description": (
"Searches the internal knowledge base for documents relevant "
"to a given query. Use this when the user needs information "
"from internal documentation, standards, or guidelines."
),
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The search query.",
},
"domain": {
"type": "string",
"enum": ["engineering", "hr", "finance", "general"],
"description": "The knowledge domain to search.",
},
},
"required": ["query"],
},
},
},
{
"type": "function",
"function": {
"name": "summarize_meeting_notes",
"description": (
"Processes and summarizes raw meeting notes, extracting "
"key decisions and action items. Use this when the user "
"provides meeting notes and wants a structured summary."
),
"parameters": {
"type": "object",
"properties": {
"notes": {
"type": "string",
"description": "The raw meeting notes text.",
},
"output_format": {
"type": "string",
"enum": ["bullet", "prose"],
"description": "The desired format for the summary.",
},
},
"required": ["notes"],
},
},
},
]
FUNCTION_REGISTRY = {
"search_knowledge_base": search_knowledge_base,
"summarize_meeting_notes": summarize_meeting_notes,
}
def run_local_agent(user_query: str, model: str = "llama4:scout") -> str:
"""
Runs the agentic loop against a locally hosted Ollama model.
The logic is structurally identical to the remote version; only the
client and model name differ.
Args:
user_query: The user's question or request.
model: The Ollama model name to use.
Returns:
The model's final natural-language response.
"""
messages = [
{
"role": "system",
"content": (
"You are a knowledgeable internal assistant. You have access "
"to the company knowledge base and can process meeting notes. "
"Always use the available tools to ground your answers in "
"real internal data."
),
},
{"role": "user", "content": user_query},
]
# First call: let the local model decide on tool usage.
# ollama.chat() returns an ollama.ChatResponse object; access fields
# via attribute notation, not dictionary subscript.
response = ollama.chat(
model=model,
messages=messages,
tools=TOOLS,
)
response_message = response.message
messages.append(
{
"role": response_message.role,
"content": response_message.content or "",
}
)
# Execute any tool calls the model requested
tool_calls = response_message.tool_calls or []
if tool_calls:
for tool_call in tool_calls:
function_name = tool_call.function.name
function_args = tool_call.function.arguments
# arguments may arrive as a dict or as a JSON string
if isinstance(function_args, str):
function_args = json.loads(function_args)
result = FUNCTION_REGISTRY[function_name](**function_args)
messages.append(
{
"role": "tool",
"content": json.dumps(result),
}
)
# Second call: synthesize results into a final answer
final_response = ollama.chat(
model=model,
messages=messages,
)
return final_response.message.content or ""
return response_message.content or ""
if __name__ == "__main__":
query = (
"What are our internal standards for PLC programming? "
"Also, here are some meeting notes: 'We discussed the prototype "
"deadline and agreed engineering needs to deliver by sprint end. "
"PM will update the dashboard. Legal needs to review the vendor "
"contract.' Can you summarize those as bullet points?"
)
print(f"User: {query}\n")
print(f"Local Agent: {run_local_agent(query)}")
What is remarkable about these two examples side by side is how similar they are. The tool definitions are identical. The function implementations are identical. The agentic loop logic is nearly identical. The only meaningful differences are the client library being used and the model name. This is not an accident. It reflects a deliberate convergence in the industry around a common interface standard for tool-calling, which makes it genuinely practical to build systems that can switch between local and remote models depending on the task, the data sensitivity, or the cost constraints of the moment.
A common pattern in production systems is to use a local model for tasks that involve sensitive internal data, and a more powerful remote model for tasks that require deep reasoning over public information. The routing logic between these two modes can itself be handled by an agent — which is exactly what Chapter 8 demonstrates.
Chapter 4: The Memory Problem and How the Field Solved It
Memory is the difference between a tool and a colleague. A tool does what you tell it to do right now. A colleague remembers what you told them last month, understands how it relates to what you are telling them today, and brings that accumulated context to bear without being asked. For years, LLMs were firmly in the tool category. Every conversation started from scratch. The model had no idea who you were, what you were working on, or what you had discussed before.
This was not just an inconvenience. It was a fundamental architectural limitation. The model's "memory" was bounded by its context window, and the context window was small. Once the conversation exceeded the window size, older information was simply dropped. The model forgot.
The field attacked this problem from several directions simultaneously, and by 2025 and into 2026, a clear architecture has emerged. It is called the three-tier memory hierarchy, and it maps surprisingly well onto how human memory actually works.
Think of the innermost tier as working memory — the model's active context window, everything it can "see" right now. It is fast, immediately accessible, and ephemeral. When the conversation ends, this memory is gone. This is the same as human working memory: the things you are actively thinking about right now.
Surrounding that is a middle tier of session-scoped compressed memory. As a conversation grows long, a background process periodically summarizes the earlier parts and stores those summaries in a compressed form. When the model needs to reference earlier context, it retrieves the summaries rather than the full original text — much like how you might not remember the exact words of a meeting from last week, but you remember the key decisions clearly.
The outermost tier is long-term persistent storage: a database — typically a vector database — that stores information across sessions. User preferences, project context, learned workflows, factual knowledge accumulated over time — all of this lives in the persistent store. When the agent starts a new session, it queries this store and retrieves the most relevant memories to prime its context. This is genuine long-term memory: the things you remember from months or years ago.
The technology that makes this outermost tier work is vector embeddings and vector search. Every piece of information stored in the long-term memory is converted into a high-dimensional numerical vector that captures its semantic meaning. When the agent needs to retrieve relevant memories, it converts the current query into a vector and finds the stored memories whose vectors are closest to the query vector. This is called semantic search, and it is far more powerful than keyword search because it finds conceptually related information even when the exact words do not match.
The following example demonstrates a complete long-term memory system built with ChromaDB as the vector store. It shows how an agent can store memories after each interaction and retrieve relevant ones at the start of the next interaction — the pattern that makes an agent feel like it truly knows you over time.
Install dependencies:
pip install chromadb>=1.5.9 openai>=3.0.0 sentence-transformers>=5.5.1
Run:
python agent_memory.py
"""
agent_memory.py
A complete long-term memory system for AI agents using ChromaDB as the
vector store. Demonstrates memory storage, retrieval, and integration
into the agent's context window.
This pattern gives agents persistent, cross-session memory that feels
natural and human-like to the end user.
Tested with: chromadb>=1.5.9, openai>=3.0.0,
sentence-transformers>=5.5.1, Python 3.10+
"""
import json
import os
import uuid
from datetime import datetime
from typing import Optional
import chromadb
from chromadb.utils import embedding_functions
from openai import OpenAI
# ---------------------------------------------------------------------------
# Memory system configuration. These constants control the behavior of the
# memory retrieval system. Tuning these values affects the balance between
# recall breadth and context window efficiency.
# ---------------------------------------------------------------------------
MEMORY_COLLECTION_NAME = "agent_long_term_memory"
MAX_MEMORIES_TO_RETRIEVE = 5 # How many past memories to inject per session
MEMORY_RELEVANCE_THRESHOLD = 0.7 # Cosine similarity threshold for retrieval
class AgentMemorySystem:
"""
Manages long-term persistent memory for an AI agent using ChromaDB.
This class handles three core operations:
1. Storing new memories after each interaction.
2. Retrieving relevant memories at the start of each new session.
3. Summarizing and compressing old memories to prevent storage bloat.
"""
def __init__(self, user_id: str, persist_directory: str = "./agent_memory") -> None:
"""
Initializes the memory system for a specific user.
Args:
user_id: Unique identifier for the user. Memories are
namespaced per user for privacy and relevance.
persist_directory: Directory where ChromaDB will persist data to
disk across process restarts.
"""
self.user_id = user_id
# Initialize ChromaDB with disk persistence so memories survive
# application restarts. This is what makes it truly "long-term."
self.chroma_client = chromadb.PersistentClient(path=persist_directory)
# Use a sentence-transformer model for generating embeddings locally.
# This avoids sending memory content to external embedding APIs.
self.embedding_function = (
embedding_functions.SentenceTransformerEmbeddingFunction(
model_name="all-MiniLM-L6-v2"
)
)
# Get or create the memory collection for this application
self.collection = self.chroma_client.get_or_create_collection(
name=MEMORY_COLLECTION_NAME,
embedding_function=self.embedding_function,
metadata={"hnsw:space": "cosine"},
)
def store_memory(
self,
content: str,
memory_type: str = "interaction",
importance: float = 0.5,
) -> str:
"""
Stores a new memory in the persistent vector store.
Args:
content: The text content of the memory to store.
memory_type: Category of memory ('interaction', 'preference',
'fact', 'workflow').
importance: A score from 0.0 to 1.0 indicating how important
this memory is. Higher-importance memories are
weighted more heavily during retrieval.
Returns:
The unique ID assigned to this memory.
"""
memory_id = str(uuid.uuid4())
timestamp = datetime.utcnow().isoformat()
self.collection.add(
ids=[memory_id],
documents=[content],
metadatas=[
{
"user_id": self.user_id,
"memory_type": memory_type,
"importance": importance,
"timestamp": timestamp,
}
],
)
print(f"[Memory] Stored: '{content[:60]}...' (ID: {memory_id[:8]})")
return memory_id
def retrieve_relevant_memories(
self,
query: str,
n_results: int = MAX_MEMORIES_TO_RETRIEVE,
) -> list[dict]:
"""
Retrieves the most semantically relevant memories for a given query.
This is the core of the memory system. The query is embedded into a
vector, and ChromaDB finds the stored memories whose vectors are
closest in semantic space. The result is a ranked list of memories
that are contextually relevant to the current conversation.
Args:
query: The current user query or conversation context.
n_results: Maximum number of memories to retrieve.
Returns:
A list of memory dictionaries, each containing the content,
metadata, and relevance score.
"""
# Guard: if the collection is empty, return early to avoid a
# ChromaDB error when n_results exceeds the number of stored docs.
try:
total_in_collection = self.collection.count()
except Exception:
return []
if total_in_collection == 0:
return []
safe_n = min(n_results, total_in_collection)
try:
results = self.collection.query(
query_texts=[query],
n_results=safe_n,
where={"user_id": self.user_id},
)
except Exception:
# If the where-filter reduces results below safe_n ChromaDB
# may still raise; return empty rather than crash.
return []
memories: list[dict] = []
if results["documents"] and results["documents"][0]:
for doc, meta, dist in zip(
results["documents"][0],
results["metadatas"][0],
results["distances"][0],
):
# Convert distance to similarity.
# For cosine space: similarity = 1 - distance.
similarity = 1.0 - dist
if similarity >= MEMORY_RELEVANCE_THRESHOLD:
memories.append(
{
"content": doc,
"type": meta.get("memory_type", "unknown"),
"timestamp": meta.get("timestamp", "unknown"),
"importance": meta.get("importance", 0.5),
"relevance": round(similarity, 3),
}
)
# Sort by a combined score of relevance and importance
memories.sort(
key=lambda m: m["relevance"] * 0.7 + m["importance"] * 0.3,
reverse=True,
)
return memories
def format_memories_for_context(self, memories: list[dict]) -> str:
"""
Formats retrieved memories into a string suitable for injection
into the model's system prompt.
Args:
memories: List of memory dictionaries from retrieve_relevant_memories.
Returns:
A formatted string ready to be inserted into the system prompt.
"""
if not memories:
return "No relevant memories found for this session."
lines = ["RELEVANT MEMORIES FROM PREVIOUS SESSIONS:", "-" * 40]
for i, mem in enumerate(memories, 1):
lines.append(
f"{i}. [{mem['type'].upper()}] (Relevance: {mem['relevance']}) "
f"{mem['content']}"
)
lines.append("-" * 40)
return "\n".join(lines)
class MemoryAwareAgent:
"""
An AI agent that uses the AgentMemorySystem to maintain persistent,
cross-session memory. Each interaction both reads from and writes to
the long-term memory store.
"""
def __init__(self, user_id: str) -> None:
"""
Args:
user_id: Unique identifier for the user this agent serves.
"""
self.user_id = user_id
self.memory = AgentMemorySystem(user_id=user_id)
self.openai_client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))
def chat(self, user_message: str) -> str:
"""
Processes a user message with full memory integration.
The flow is:
1. Retrieve relevant memories from long-term storage.
2. Inject memories into the system prompt.
3. Send the enriched prompt to the model.
4. Store the interaction as a new memory.
5. Return the model's response.
Args:
user_message: The user's input message.
Returns:
The agent's response as a string.
"""
# Step 1: Retrieve memories relevant to the current query
relevant_memories = self.memory.retrieve_relevant_memories(user_message)
memory_context = self.memory.format_memories_for_context(relevant_memories)
print(f"\n[Memory] Retrieved {len(relevant_memories)} relevant memories.")
# Step 2: Build the system prompt with memory context injected
system_prompt = (
"You are a personalized AI assistant with persistent memory. "
"You remember past interactions with this user and use that context "
"to provide more relevant and personalized responses.\n\n"
f"{memory_context}\n\n"
"Use the above memories to inform your response when relevant. Do not "
"explicitly list the memories back to the user unless asked; instead, "
"naturally incorporate the context they provide."
)
# Step 3: Call the model with the memory-enriched context
response = self.openai_client.chat.completions.create(
model="gpt-5", # Current flagship model, August 2025+
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_message},
],
)
agent_response = response.choices[0].message.content
# Step 4: Store this interaction as a new memory for future sessions.
# We store a concise summary of what was discussed, not the full text.
interaction_summary = (
f"User asked about: {user_message[:100]}. "
f"Agent responded with: {agent_response[:150]}"
)
self.memory.store_memory(
content=interaction_summary,
memory_type="interaction",
importance=0.6,
)
# Store explicit preferences or facts if detected
preference_triggers = [
"i prefer", "i like", "i always", "i usually", "my name is"
]
if any(phrase in user_message.lower() for phrase in preference_triggers):
self.memory.store_memory(
content=f"User preference/fact: {user_message}",
memory_type="preference",
importance=0.9, # Preferences are highly important to remember
)
return agent_response
if __name__ == "__main__":
# Simulate two separate sessions to demonstrate cross-session memory
print("=" * 60)
print("SESSION 1: First interaction with the agent")
print("=" * 60)
agent = MemoryAwareAgent(user_id="employee_42")
response1 = agent.chat(
"My name is Maria and I prefer concise bullet-point summaries. "
"I am working on the digital twin project for the Munich plant."
)
print(f"\nAgent: {response1}")
response2 = agent.chat(
"What are the key challenges in industrial digital twin implementations?"
)
print(f"\nAgent: {response2}")
print("\n" + "=" * 60)
print("SESSION 2: New session — agent should remember Maria")
print("=" * 60)
# Create a fresh agent instance (simulating a new session / process restart).
# The memory persists because ChromaDB saved it to disk.
agent_session2 = MemoryAwareAgent(user_id="employee_42")
response3 = agent_session2.chat(
"Can you give me an update on what we discussed last time?"
)
print(f"\nAgent: {response3}")
The memory system above is doing something that would have seemed like science fiction just a few years ago. It is giving an AI agent a genuine sense of continuity across time. Maria comes back the next day, and the agent already knows she prefers bullet points, that she is working on the Munich plant project, and what they discussed before. She does not have to re-introduce herself. She does not have to re-explain her context. The agent picks up where they left off, exactly as a good human colleague would.
The vector database is the key enabler here. It stores memories not as exact text matches but as semantic representations, which means the agent can find relevant memories even when the current query uses completely different words than the stored memory. If Maria asks about "automation challenges" and the stored memory mentions "digital twin implementation difficulties," the semantic similarity between those concepts will surface the memory correctly.
This is not magic. It is mathematics — specifically the geometry of high-dimensional vector spaces, where semantically similar concepts cluster together. But from the user's perspective, it feels like the agent genuinely understands and remembers them.
Chapter 5: The Context Window Revolution — When More Is Genuinely More
While memory systems solve the problem of cross-session continuity, context windows solve a different but equally important problem: how much information can the model hold in its "mind" at once during a single session.
The numbers here are genuinely staggering when you put them in human terms. A context window of one million tokens — which Gemini 2.5 Pro offered in early 2025 — can hold approximately 750,000 words of text. That is roughly equivalent to ten full-length novels, or the complete source code of a medium-sized software project, or five years of email correspondence. Meta's Llama 4 Scout pushed this to ten million tokens in April 2025. By August 2026, leading frontier models routinely offer context windows exceeding twenty million tokens, and specialized research models have demonstrated hundred-million-token contexts in controlled settings. All of it is available to the model simultaneously, in a single pass, without any retrieval or summarization needed.
Why does this matter so much? Because retrieval is lossy. Every time you use RAG or memory retrieval to pull relevant information into the context, you are making a bet that your retrieval system found the right information. Sometimes it does not. Sometimes the most important piece of context is in a document that did not score highly enough to be retrieved. With a large enough context window, you can simply load everything and let the model figure out what is relevant. No retrieval system, no lossy compression, no missed connections.
That said, large context windows and memory systems are not competitors — they are complements. The context window handles the current session's working set. The memory system handles cross-session continuity. The best production systems use both, loading relevant long-term memories into the context window at the start of each session, and then using the full context window to reason over the current session's documents and data.
The figure below illustrates the relationship between these memory tiers:
MEMORY ARCHITECTURE OVERVIEW
TIER 3: Long-Term Persistent Storage (Cross-Session)
+----------------------------------------------------------+
| Vector Database (ChromaDB / Qdrant / Pinecone) |
| - User preferences and facts |
| - Past interaction summaries |
| - Learned workflows and domain knowledge |
| - Persists indefinitely across all sessions |
+----------------------------------------------------------+
|
[Semantic Retrieval]
Top-K most relevant
memories injected
|
v
TIER 2: Session-Scoped Compressed Memory
+----------------------------------------------------------+
| Conversation Summaries (in-memory or short-term DB) |
| - Compressed summaries of earlier conversation turns |
| - Refreshed each session |
| - Prevents context window overflow during long sessions |
+----------------------------------------------------------+
|
[Injected into context]
|
v
TIER 1: In-Context Working Memory (Active Session)
+----------------------------------------------------------+
| Active Context Window (20M+ tokens, August 2026) |
| - Current conversation turns |
| - Retrieved long-term memories |
| - Loaded documents and data |
| - Tool call results |
| - System instructions |
+----------------------------------------------------------+
|
v
+--------------------+
| LLM Reasoning |
| Core (GPT-5.6, |
| Llama 4 Scout, |
| etc.) |
+--------------------+
The practical implication of massive context windows for knowledge workers is enormous. Consider a lawyer who needs to analyze a complex contract against a library of precedents. With a twenty-million-token context window, they can load the entire contract, all relevant precedents, and their firm's internal guidelines into a single prompt and ask the model to identify conflicts, missing clauses, and risk factors. No chunking, no retrieval, no hoping that the right precedent gets surfaced. Just load everything and ask.
Or consider a software engineer who needs to understand a large, unfamiliar codebase. With a ten-million-token context window, they can load the entire repository and ask the model to explain the architecture, trace a specific execution path, or identify all the places where a particular function is called. The model can see the whole system at once, not just fragments.
This is a qualitative change in capability, not just a quantitative one. It changes what kinds of questions you can ask and what kinds of answers you can get.
Chapter 6: The Rise of Agent Teams — Specialization and Orchestration
Here is where things get genuinely exciting, and also genuinely complex. A single agent, no matter how capable, has limits. It can only do one thing at a time. It can only hold so much context. It can only be optimized for one kind of task at a time. These limits are not just technical inconveniences; they reflect something fundamental about the nature of complex work.
Complex work is inherently parallel and inherently specialized. When a company launches a new product, the marketing team, the legal team, the engineering team, and the finance team all work simultaneously, each bringing deep expertise to their domain. No single person does all of that work. No single person could.
The same insight applies to AI agents. Instead of building one general-purpose agent and hoping it can do everything, the field has converged on building teams of specialized agents that collaborate. One agent specializes in research and information retrieval. Another specializes in code generation and testing. Another specializes in document writing and editing. Another specializes in data analysis and visualization. An orchestrator agent coordinates their work, delegates tasks, and synthesizes their outputs into a coherent result.
This is the multi-agent paradigm, and it is producing results that are dramatically better than single-agent approaches. Research through 2025 and into 2026 has found that multi-agent architectures process complex tasks 50 to 60 percent more efficiently than single-model approaches, and achieve success rates exceeding 90 percent in structured workflows like smart manufacturing.
There are several standard orchestration patterns that have emerged as the field has matured. Sequential orchestration runs agents one after another, with the output of each agent feeding into the next — simple and effective for linear workflows like research, then writing, then editing. Parallel orchestration runs multiple agents simultaneously on independent subtasks and aggregates their results, which is efficient for tasks that can be decomposed into independent pieces. Hierarchical orchestration uses a supervisor agent to analyze the overall goal, delegate subtasks to worker agents, and synthesize their results — the most flexible pattern, and now the production default for complex workflows.
The following example demonstrates a multi-agent system built with CrewAI, one of the leading frameworks for multi-agent orchestration. The system assembles a team of three specialized agents to produce a comprehensive technical analysis report, with each agent contributing its area of expertise.
Install dependencies:
pip install "crewai[tools]>=1.15.15" openai>=3.0.0
Run:
python multi_agent_crew.py
"""
multi_agent_crew.py
A multi-agent system using CrewAI to produce a comprehensive technical
analysis report. Three specialized agents collaborate sequentially:
1. Research Agent: Gathers and synthesizes technical information.
2. Analysis Agent: Performs deep analysis of the research findings.
3. Writing Agent: Produces a polished, structured report.
This demonstrates how specialization and orchestration combine to produce
results that exceed what any single agent could achieve alone.
Tested with: crewai[tools]>=1.15.15, openai>=3.0.0, Python 3.10+
"""
import os
from crewai import Agent, Crew, Process, Task
from crewai.tools import BaseTool
from pydantic import BaseModel, Field
# ---------------------------------------------------------------------------
# Custom tool definitions. CrewAI agents are equipped with tools that extend
# their capabilities beyond pure language generation. Tools are the "hands"
# that let agents interact with the world.
# ---------------------------------------------------------------------------
class TechnicalDatabaseInput(BaseModel):
"""Input schema for the TechnicalDatabaseTool."""
topic: str = Field(description="The technical topic to look up.")
depth: str = Field(
default="overview",
description="Depth of information: 'overview', 'detailed', or 'expert'.",
)
class TechnicalDatabaseTool(BaseTool):
"""
Simulates querying a technical knowledge database.
In production, this would connect to a real technical documentation
system, a vector database, or an internal wiki.
"""
name: str = "technical_database"
description: str = (
"Queries the internal technical knowledge database for information "
"on engineering topics, standards, and best practices."
)
args_schema: type[BaseModel] = TechnicalDatabaseInput
def _run(self, topic: str, depth: str = "overview") -> str:
"""Executes the database query and returns formatted results."""
database = {
"digital twin": {
"overview": (
"A digital twin is a virtual representation of a physical "
"asset, process, or system. It synchronizes with its "
"physical counterpart in real-time using sensor data and "
"IoT connectivity. Key applications include predictive "
"maintenance, process optimization, and virtual testing."
),
"detailed": (
"Digital twins in industrial settings leverage OPC-UA for "
"data acquisition, time-series databases for historical "
"data, and physics-based simulation engines for predictive "
"modeling. The Industrial Metaverse platform integrates "
"digital twins with collaborative 3D environments."
),
},
"edge computing": {
"overview": (
"Edge computing processes data near the source of generation "
"rather than in a centralized cloud. This reduces latency, "
"bandwidth consumption, and dependency on cloud connectivity."
),
"detailed": (
"Industrial edge computing uses ruggedized industrial IPC "
"hardware for reliable on-site processing. Key protocols "
"include MQTT for telemetry and OPC-UA for machine "
"communication. Edge AI enables real-time inference for "
"quality control and anomaly detection without cloud "
"round-trips."
),
},
}
topic_lower = topic.lower()
for key in database:
if key in topic_lower:
return database[key].get(depth, database[key]["overview"])
return (
f"No detailed records found for '{topic}'. "
"General search returned no results."
)
# ---------------------------------------------------------------------------
# Agent factory functions. Each agent has a carefully crafted role, goal,
# and backstory. These are not just labels; they shape the model's behavior
# and the quality of its output. Think of them as the agent's professional
# identity and motivation.
# ---------------------------------------------------------------------------
def create_research_agent(llm_model: str = "gpt-5") -> Agent:
"""
Creates a specialized research agent equipped with database access.
The research agent's job is to gather comprehensive, accurate information
on the assigned topic. It does not analyze or write; it researches.
This focus is what makes it effective.
"""
return Agent(
role="Senior Technical Research Analyst",
goal=(
"Gather comprehensive, accurate, and well-sourced technical "
"information on the assigned topic. Identify key concepts, "
"current state of the art, challenges, and opportunities."
),
backstory=(
"You are a seasoned technical researcher with 15 years of "
"experience in industrial automation and digital transformation. "
"You have a talent for finding the most relevant information "
"quickly and presenting it in a structured, factual manner. "
"You never speculate; you only report what you can verify."
),
tools=[TechnicalDatabaseTool()],
llm=llm_model,
verbose=True,
allow_delegation=False,
)
def create_analysis_agent(llm_model: str = "gpt-5") -> Agent:
"""
Creates a specialized analysis agent that processes research findings.
The analysis agent receives the researcher's output and performs deep
analytical work: identifying patterns, drawing conclusions, assessing
risks, and generating strategic insights.
"""
return Agent(
role="Principal Systems Analyst",
goal=(
"Perform deep analysis of the research findings. Identify "
"strategic implications, technical risks, implementation "
"challenges, and concrete recommendations. Produce insights "
"that go beyond the surface level of the raw research."
),
backstory=(
"You are a principal systems analyst with deep expertise in "
"translating technical research into actionable strategic "
"insights. You think in systems, always considering second-order "
"effects and interdependencies. Your analyses are known for "
"their clarity, depth, and practical relevance."
),
llm=llm_model,
verbose=True,
allow_delegation=False,
)
def create_writing_agent(llm_model: str = "gpt-5") -> Agent:
"""
Creates a specialized technical writing agent.
The writing agent takes the research and analysis and transforms them
into a polished, professional report. It focuses on clarity, structure,
and readability, not on generating new content.
"""
return Agent(
role="Senior Technical Writer",
goal=(
"Transform the research findings and analytical insights into "
"a clear, well-structured, and professionally written technical "
"report. The report must be accessible to both technical and "
"non-technical stakeholders."
),
backstory=(
"You are a senior technical writer who has spent a decade "
"making complex technical content accessible to diverse "
"audiences. You have a gift for structure and clarity, and "
"you take pride in producing documents that people actually "
"want to read. You never pad content; every sentence earns "
"its place."
),
llm=llm_model,
verbose=True,
allow_delegation=False,
)
def build_analysis_crew(topic: str, llm_model: str = "gpt-5") -> Crew:
"""
Assembles the complete multi-agent crew for technical analysis.
This function wires together the agents and their tasks into a
sequential pipeline. The output of each task is automatically
passed as context to the next task by CrewAI's orchestration layer.
Args:
topic: The technical topic to analyze.
llm_model: The LLM model identifier to use for all agents.
Returns:
A configured Crew ready to execute.
"""
research_agent = create_research_agent(llm_model)
analysis_agent = create_analysis_agent(llm_model)
writing_agent = create_writing_agent(llm_model)
# Task 1: Research. The agent uses its database tool to gather information.
research_task = Task(
description=(
f"Research the topic: '{topic}'. Use the technical database tool "
f"to gather both overview and detailed information. Compile your "
f"findings into a structured research brief covering: key concepts, "
f"current state of the art, main challenges, and emerging trends."
),
expected_output=(
"A structured research brief of 400-600 words covering key "
"concepts, current state of the art, main challenges, and "
"emerging trends related to the topic."
),
agent=research_agent,
)
# Task 2: Analysis. The agent receives the research brief as context.
analysis_task = Task(
description=(
f"Analyze the research brief about '{topic}' provided by the "
f"research agent. Identify: (1) the three most significant "
f"strategic implications, (2) the top implementation risks and "
f"how to mitigate them, (3) concrete recommendations for an "
f"organization considering adoption, and (4) a realistic "
f"assessment of the technology's maturity level."
),
expected_output=(
"A structured analytical report of 400-600 words covering "
"strategic implications, implementation risks, recommendations, "
"and technology maturity assessment."
),
agent=analysis_agent,
context=[research_task], # Explicitly receives research output
)
# Task 3: Writing. The agent synthesizes both previous outputs.
writing_task = Task(
description=(
f"Write a comprehensive technical report on '{topic}' based on "
f"the research brief and analytical report provided by the "
f"previous agents. The report should have: an executive summary "
f"(150 words), a technical overview section, a strategic analysis "
f"section, and a recommendations section. Use clear headings and "
f"professional language suitable for a C-suite audience."
),
expected_output=(
"A complete, professionally formatted technical report of "
"800-1000 words with clear sections: Executive Summary, "
"Technical Overview, Strategic Analysis, and Recommendations."
),
agent=writing_agent,
context=[research_task, analysis_task], # Receives both previous outputs
)
return Crew(
agents=[research_agent, analysis_agent, writing_agent],
tasks=[research_task, analysis_task, writing_task],
process=Process.sequential,
verbose=True,
)
if __name__ == "__main__":
TOPIC = "Digital Twin Technology in Industrial Manufacturing"
print(f"Launching Multi-Agent Analysis Crew for: {TOPIC}")
print("-" * 70)
crew = build_analysis_crew(topic=TOPIC, llm_model="gpt-5")
result = crew.kickoff()
print("\nFINAL REPORT")
print("-" * 70)
print(result)
What the code above is doing is genuinely fascinating when you think about it carefully. Three completely separate AI agents — each with its own identity, goal, and area of expertise — are collaborating on a single complex task. The researcher does not try to analyze. The analyst does not try to write. The writer does not try to research. Each agent stays in its lane, does its job exceptionally well, and passes its output to the next agent in the chain.
The orchestration layer — in this case CrewAI's sequential process — handles the handoffs automatically. It ensures that the analysis agent has access to the research agent's output, and that the writing agent has access to both. The agents do not need to know about each other's implementation details. They just need to know what they are supposed to produce and what context they have been given.
This is the power of specialization combined with orchestration. The whole genuinely becomes greater than the sum of its parts.
Chapter 7: Hierarchical Orchestration — The Supervisor Pattern
Sequential orchestration is elegant for linear workflows, but the real world is rarely linear. Complex tasks branch, loop, and require dynamic decision-making about which agent to engage next. This is where hierarchical orchestration — also called the supervisor pattern — becomes essential.
In the supervisor pattern, a central orchestrator agent receives the overall goal and is responsible for breaking it down into subtasks, delegating those subtasks to appropriate worker agents, evaluating the results, and deciding what to do next. The orchestrator is not just a router; it is a reasoning entity that can adapt the workflow dynamically based on what it learns from each worker agent's output.
Think of it like a skilled project manager who can read the situation, reassign work when a team member gets stuck, recognize when a deliverable needs to be revised, and synthesize disparate contributions into a coherent whole. The supervisor agent plays exactly this role.
The following example implements a supervisor pattern using LangGraph, which provides a graph-based framework for building stateful, multi-agent workflows. LangGraph is particularly powerful for hierarchical orchestration because it represents the workflow as a directed graph, where nodes are agents or processing steps and edges are the transitions between them. Conditional edges allow the graph to branch dynamically based on the supervisor's decisions.
Install dependencies:
pip install langgraph>=1.2.11 langchain-openai>=1.5.0 langchain-core
Run:
python supervisor_agent.py
"""
supervisor_agent.py
Implements the hierarchical supervisor pattern using LangGraph.
A supervisor agent dynamically routes tasks to specialized worker agents
based on the nature of each subtask, demonstrating adaptive orchestration.
This pattern is ideal for workflows where the sequence of steps is not
known in advance and must be determined dynamically.
Tested with: langgraph>=1.2.11, langchain-openai>=1.5.0, Python 3.10+
"""
import os
from typing import Annotated, Literal, TypedDict
from langchain_core.messages import AIMessage, BaseMessage, HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages
# ---------------------------------------------------------------------------
# State definition. In LangGraph, the state is the shared data structure
# that flows through the graph. Every node reads from and writes to this
# state. The 'add_messages' annotation tells LangGraph to append new
# messages to the list rather than replacing it.
# ---------------------------------------------------------------------------
class WorkflowState(TypedDict):
"""
The shared state that flows through the supervisor workflow graph.
Attributes:
messages: The full conversation history, including all agent
outputs. LangGraph appends to this list automatically.
next_agent: The supervisor's routing decision: which agent to call
next, or FINISH to terminate the workflow.
task_context: Additional context passed between agents.
"""
messages: Annotated[list[BaseMessage], add_messages]
next_agent: str
task_context: dict
# ---------------------------------------------------------------------------
# LLM factory
# ---------------------------------------------------------------------------
def create_llm(model: str = "gpt-5", temperature: float = 0.3) -> ChatOpenAI:
"""
Factory function for creating LLM instances with consistent configuration.
Args:
model: The OpenAI model identifier.
temperature: Controls output randomness. Lower values produce more
consistent, deterministic outputs.
Returns:
A configured ChatOpenAI instance.
"""
return ChatOpenAI(
model=model,
temperature=temperature,
api_key=os.environ.get("OPENAI_API_KEY"),
)
# ---------------------------------------------------------------------------
# Worker agent node functions. Each worker is a focused specialist that
# receives the current state and produces a specific type of output.
# ---------------------------------------------------------------------------
def code_review_agent(state: WorkflowState) -> dict:
"""
Specialized agent for reviewing code quality, security, and best practices.
Examines code snippets or descriptions and provides detailed feedback on
correctness, style, security vulnerabilities, and architectural concerns.
Args:
state: The current workflow state containing the task context.
Returns:
Partial state update with the code review findings appended.
"""
llm = create_llm(temperature=0.1)
system_prompt = (
"You are an expert code reviewer specializing in Python, "
"industrial automation software, and secure coding practices. "
"Review the provided code or code description and provide specific, "
"actionable feedback covering: correctness, security vulnerabilities, "
"performance issues, and adherence to clean code principles. "
"Be specific and constructive."
)
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": str(state["messages"][-1].content)},
]
response = llm.invoke(messages)
review_output = AIMessage(content=f"[CODE REVIEW AGENT]: {response.content}")
return {"messages": [review_output], "next_agent": state["next_agent"]}
def documentation_agent(state: WorkflowState) -> dict:
"""
Specialized agent for generating technical documentation.
Produces clear, comprehensive documentation including API references,
user guides, and inline code comments.
Args:
state: The current workflow state.
Returns:
Partial state update with generated documentation appended.
"""
llm = create_llm(temperature=0.4)
system_prompt = (
"You are a senior technical writer specializing in software "
"documentation. Generate clear, comprehensive, and well-structured "
"documentation for the provided code or system description. Include: "
"purpose and overview, parameter descriptions, usage examples, and "
"any important caveats or limitations. Follow Google's developer "
"documentation style guide."
)
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": str(state["messages"][-1].content)},
]
response = llm.invoke(messages)
doc_output = AIMessage(content=f"[DOCUMENTATION AGENT]: {response.content}")
return {"messages": [doc_output], "next_agent": state["next_agent"]}
def testing_agent(state: WorkflowState) -> dict:
"""
Specialized agent for generating test cases and testing strategies.
Analyzes code or functionality descriptions and produces comprehensive
test suites including unit tests, edge cases, and integration test
strategies.
Args:
state: The current workflow state.
Returns:
Partial state update with test cases and strategy appended.
"""
llm = create_llm(temperature=0.2)
system_prompt = (
"You are a senior QA engineer and testing specialist. "
"Generate comprehensive test cases for the provided code or "
"functionality. Include: unit tests using pytest, edge case "
"identification, boundary value analysis, and integration test "
"strategy. Write actual test code where possible, not just descriptions."
)
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": str(state["messages"][-1].content)},
]
response = llm.invoke(messages)
test_output = AIMessage(content=f"[TESTING AGENT]: {response.content}")
return {"messages": [test_output], "next_agent": state["next_agent"]}
def supervisor_agent(state: WorkflowState) -> dict:
"""
The central orchestrator that routes tasks to appropriate worker agents.
Reads the current state, evaluates what has been done, and decides what
to do next. It can route to any worker agent or terminate the workflow
when the goal is achieved.
Args:
state: The current workflow state.
Returns:
Partial state update with the supervisor's routing decision.
"""
llm = create_llm(temperature=0.1)
# Build a summary of what has been accomplished so far
completed_work: list[str] = []
for msg in state["messages"]:
if isinstance(msg, AIMessage) and msg.content.startswith("["):
agent_name = msg.content.split("]")[0].replace("[", "")
completed_work.append(agent_name)
completed_summary = (
f"Completed: {', '.join(completed_work)}"
if completed_work
else "No work completed yet."
)
system_prompt = (
"You are a workflow supervisor managing a software development team. "
"Your job is to route the current task to the most appropriate "
"specialist agent, or to finish the workflow if all necessary work "
"is complete.\n\n"
"Available agents:\n"
"- code_reviewer: Reviews code quality, security, and best practices\n"
"- documentation_writer: Generates technical documentation\n"
"- test_engineer: Creates test cases and testing strategies\n\n"
f"{completed_summary}\n\n"
"Based on the task and what has been completed, respond with EXACTLY ONE of:\n"
'- "code_reviewer" (if code review is needed and not done)\n'
'- "documentation_writer" (if documentation is needed and not done)\n'
'- "test_engineer" (if testing strategy is needed and not done)\n'
'- "FINISH" (if all necessary work is complete)\n\n'
"Respond with only the agent name or FINISH. Nothing else."
)
messages_for_supervisor = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": str(state["messages"][0].content)},
]
response = llm.invoke(messages_for_supervisor)
routing_decision = response.content.strip().lower()
routing_map = {
"code_reviewer": "code_reviewer",
"documentation_writer": "documentation_writer",
"test_engineer": "test_engineer",
"finish": "FINISH",
}
next_agent = routing_map.get(routing_decision, "FINISH")
return {
"messages": [],
"next_agent": next_agent,
"task_context": state.get("task_context", {}),
}
def route_from_supervisor(
state: WorkflowState,
) -> Literal["code_reviewer", "documentation_writer", "test_engineer", "__end__"]:
"""
Conditional edge function that reads the supervisor's routing decision
and returns the name of the next node to execute.
LangGraph uses this function to determine which edge to follow in the
graph after the supervisor node executes.
Args:
state: The current workflow state containing next_agent.
Returns:
The name of the next node, or '__end__' to terminate the graph.
"""
next_agent = state.get("next_agent", "FINISH")
if next_agent == "FINISH":
return END # END == "__end__" in LangGraph
return next_agent # type: ignore[return-value]
def build_supervisor_graph() -> StateGraph:
"""
Constructs the LangGraph workflow graph for the supervisor pattern.
Graph structure:
START -> supervisor -> [code_reviewer | documentation_writer |
test_engineer | END]
Each worker -> supervisor (loop back for next routing decision)
Returns:
A compiled LangGraph StateGraph ready for execution.
"""
graph = StateGraph(WorkflowState)
# Add all nodes to the graph
graph.add_node("supervisor", supervisor_agent)
graph.add_node("code_reviewer", code_review_agent)
graph.add_node("documentation_writer", documentation_agent)
graph.add_node("test_engineer", testing_agent)
# The workflow always starts with the supervisor
graph.add_edge(START, "supervisor")
# The supervisor uses a conditional edge to route dynamically
graph.add_conditional_edges(
"supervisor",
route_from_supervisor,
{
"code_reviewer": "code_reviewer",
"documentation_writer": "documentation_writer",
"test_engineer": "test_engineer",
END: END,
},
)
# After each worker completes, control returns to the supervisor.
# This creates the feedback loop that enables adaptive orchestration.
graph.add_edge("code_reviewer", "supervisor")
graph.add_edge("documentation_writer", "supervisor")
graph.add_edge("test_engineer", "supervisor")
return graph.compile()
if __name__ == "__main__":
workflow = build_supervisor_graph()
initial_task = HumanMessage(
content=(
"I have a Python function that reads sensor data from an OPC-UA "
"server and stores it in a time-series database. The function "
"handles connection errors with a retry loop. Please review the "
"code quality, generate documentation, and create a testing "
"strategy for it."
)
)
initial_state: WorkflowState = {
"messages": [initial_task],
"next_agent": "",
"task_context": {},
}
print("Starting Supervisor-Orchestrated Workflow...")
print("-" * 60)
final_state = workflow.invoke(initial_state)
print("\nFINAL WORKFLOW OUTPUTS:")
print("-" * 60)
for message in final_state["messages"]:
if isinstance(message, AIMessage):
print(f"\n{message.content[:500]}...")
print("-" * 40)
The LangGraph supervisor pattern is doing something that is architecturally very different from the sequential CrewAI example. In the sequential pattern, the order of agents is fixed at design time. In the supervisor pattern, the order is determined at runtime by the supervisor's reasoning. The supervisor looks at what has been done, what still needs to be done, and what the task requires, and it makes a fresh routing decision after each worker completes.
This creates a feedback loop that is genuinely adaptive. If a worker agent produces output that reveals a new requirement, the supervisor can route to an additional agent that was not originally planned. If a task turns out to be simpler than expected, the supervisor can skip agents that are not needed. The workflow is not a rigid pipeline; it is a living, reasoning process.
The graph structure that LangGraph uses to represent this is elegant. Nodes are agents or processing steps. Edges are the transitions between them. Conditional edges implement the routing logic. The cycle from each worker back to the supervisor is what creates the adaptive loop. And the END node is the termination condition that the supervisor triggers when the goal is achieved.
Chapter 8: Putting It All Together — A Unified Agentic Architecture
We have now seen all the major components of next-generation agentic AI systems: tool calling, local and remote model integration, long-term memory with vector databases, massive context windows, and multi-agent orchestration patterns. The natural question is: how do these components fit together into a coherent, production-ready system?
The answer is that they form layers, each layer building on the one below it and adding a new dimension of capability. The following diagram illustrates the complete architecture of a production agentic system:
UNIFIED AGENTIC SYSTEM ARCHITECTURE
USER INTERFACE LAYER
+----------------------------------------------------------+
| Chat UI / API Endpoint / Voice Interface / Email |
| Receives user input, displays agent responses |
+----------------------------------------------------------+
|
v
ORCHESTRATION LAYER
+----------------------------------------------------------+
| Supervisor Agent (LangGraph / CrewAI / Custom) |
| - Parses user intent |
| - Routes to appropriate specialist agents |
| - Synthesizes results into coherent responses |
| - Manages workflow state and error recovery |
+----------------------------------------------------------+
| | |
v v v
SPECIALIST AGENT LAYER
+------------+ +------------+ +------------+
| Research | | Code Gen | | Analysis |
| Agent | | Agent | | Agent |
| (GPT-5) | | (Local | | (GPT-5) |
| | | Llama 4 | | |
| | | Scout) | | |
+------------+ +------------+ +------------+
| | |
v v v
TOOL LAYER
+----------------------------------------------------------+
| Web Search | Code Executor | Database Query | Email API |
| File System | Calendar | Internal APIs | Vector Search |
+----------------------------------------------------------+
|
v
MEMORY LAYER
+----------------------------------------------------------+
| In-Context Working Memory (Active Context Window) |
| Session Compressed Memory (Conversation Summaries) |
| Long-Term Persistent Memory (ChromaDB / Qdrant) |
+----------------------------------------------------------+
|
v
MODEL LAYER
+----------------------------------------------------------+
| Remote Models: GPT-5.6 (Sol, Terra, Luna), Claude,Gemini|
| Local Models: Llama 4 Scout, Phi-4-mini (via Ollama) |
| Routing logic selects model based on task and privacy |
+----------------------------------------------------------+
The routing logic at the model layer is worth dwelling on for a moment. In a sophisticated production system, not every task goes to the same model. A task involving sensitive internal data might be routed to a local Ollama model to ensure data never leaves the premises. A task requiring deep reasoning over complex technical content might be routed to GPT-5.6 or the latest Claude model like Fable 5 or Opus 5. A task requiring multimodal understanding of images or diagrams might be routed to a vision-capable model. A simple, high-volume classification task might be routed to a small, fast, cheap local model.
This intelligent model routing is itself an agentic capability. A routing agent can analyze the incoming task, assess its requirements for capability, privacy, latency, and cost, and select the appropriate model. The user never sees this routing logic. They just get the best possible answer, produced by the best possible model for that specific task.
The following example demonstrates a simple but effective model router that makes these decisions programmatically, combining both local and remote model access in a single unified interface.
Install dependencies:
pip install openai>=3.0.0 ollama>=0.32.0
Run:
python model_router.py
"""
model_router.py
An intelligent model router that selects the optimal LLM for each task
based on task characteristics: sensitivity, complexity, and latency needs.
This is the "traffic controller" of a production agentic system, ensuring
that each task is handled by the most appropriate model while respecting
privacy, cost, and performance constraints.
Prerequisites:
- Ollama running locally: ollama serve
- Local models pulled: ollama pull llama4:scout && ollama pull phi4-mini
- OpenAI API key set: export OPENAI_API_KEY="your-key-here"
Tested with: openai>=3.0.0, ollama>=0.32.0, Python 3.10+
"""
import os
from dataclasses import dataclass
from enum import Enum
from typing import Optional
import ollama
from openai import OpenAI
class ModelTier(Enum):
"""
Defines the tiers of model capability and privacy.
LOCAL_FAST: Small local model for simple, high-volume tasks.
No data leaves the machine. Very low latency.
LOCAL_CAPABLE: Larger local model for complex tasks with sensitive data.
No data leaves the machine. Moderate latency.
REMOTE_CAPABLE: Cloud model for complex reasoning with non-sensitive data.
High capability. Moderate latency and cost.
REMOTE_PREMIUM: Most capable cloud model for the hardest reasoning tasks.
Highest capability, cost, and latency.
"""
LOCAL_FAST = "local_fast"
LOCAL_CAPABLE = "local_capable"
REMOTE_CAPABLE = "remote_capable"
REMOTE_PREMIUM = "remote_premium"
@dataclass
class TaskProfile:
"""
Describes the characteristics of a task for routing purposes.
Attributes:
contains_sensitive_data: If True, the task must be routed to a
local model to prevent data exfiltration.
complexity_score: A score from 1 (trivial) to 10 (extremely
complex) estimating reasoning requirements.
requires_low_latency: If True, prefer faster models even at some
cost to capability.
task_description: A brief description for logging purposes.
"""
contains_sensitive_data: bool
complexity_score: int # 1 (trivial) to 10 (extremely complex)
requires_low_latency: bool
task_description: str
# Model configuration registry — current as of August 2026
MODEL_CONFIG: dict[ModelTier, dict] = {
ModelTier.LOCAL_FAST: {
"provider": "ollama",
"model": "phi4-mini", # 3.8B, fast, supports function calling
"description": "Local Phi-4-mini: fast, private, suitable for simple tasks",
},
ModelTier.LOCAL_CAPABLE: {
"provider": "ollama",
"model": "llama4:scout", # 17B active params, 10M token context
"description": "Local Llama 4 Scout: capable, private, for complex tasks",
},
ModelTier.REMOTE_CAPABLE: {
"provider": "openai",
"model": "gpt-5-6-sol", # Flagship model, August 2025+
"description": "GPT-5-6-Sol: Frontier model, non-sensitive tasks",
},
ModelTier.REMOTE_PREMIUM: {
"provider": "openai",
"model": "gpt-5-6-luna", # Flagship, August 2025+
"description": "GPT-5-6-Luna: Frontier model with reasonable costs",
},
}
class IntelligentModelRouter:
"""
Routes tasks to the optimal LLM based on task characteristics.
The routing logic prioritizes privacy first (sensitive data always
stays local), then balances capability against latency and cost.
"""
def __init__(self) -> None:
"""Initializes the router with both local and remote clients."""
self.openai_client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))
def _select_model_tier(self, profile: TaskProfile) -> ModelTier:
"""
Applies routing logic to select the appropriate model tier.
Decision tree:
1. Sensitive data -> local models only
a. complexity > 6 -> LOCAL_CAPABLE
b. otherwise -> LOCAL_FAST
2. Non-sensitive data:
a. low latency AND complexity <= 4 -> LOCAL_FAST
b. complexity > 7 -> REMOTE_PREMIUM
c. otherwise -> REMOTE_CAPABLE
Args:
profile: The task profile describing routing requirements.
Returns:
The selected ModelTier enum value.
"""
if profile.contains_sensitive_data:
return (
ModelTier.LOCAL_CAPABLE
if profile.complexity_score > 6
else ModelTier.LOCAL_FAST
)
if profile.requires_low_latency and profile.complexity_score <= 4:
return ModelTier.LOCAL_FAST
if profile.complexity_score > 7:
return ModelTier.REMOTE_PREMIUM
return ModelTier.REMOTE_CAPABLE
def _call_local_model(
self,
model_name: str,
messages: list[dict],
system_prompt: Optional[str] = None,
) -> str:
"""
Calls a local Ollama model with the given messages.
Args:
model_name: The Ollama model identifier.
messages: The conversation messages.
system_prompt: Optional system prompt to prepend.
Returns:
The model's response as a string.
"""
ollama_messages: list[dict] = []
if system_prompt:
ollama_messages.append({"role": "system", "content": system_prompt})
ollama_messages.extend(messages)
# ollama.chat() returns a ChatResponse; use attribute access
response = ollama.chat(model=model_name, messages=ollama_messages)
return response.message.content or ""
def _call_remote_model(
self,
model_name: str,
messages: list[dict],
system_prompt: Optional[str] = None,
) -> str:
"""
Calls a remote OpenAI model with the given messages.
Args:
model_name: The OpenAI model identifier.
messages: The conversation messages.
system_prompt: Optional system prompt to prepend.
Returns:
The model's response as a string.
"""
openai_messages: list[dict] = []
if system_prompt:
openai_messages.append({"role": "system", "content": system_prompt})
openai_messages.extend(messages)
response = self.openai_client.chat.completions.create(
model=model_name,
messages=openai_messages,
)
return response.choices[0].message.content or ""
def route_and_execute(
self,
user_message: str,
task_profile: TaskProfile,
system_prompt: Optional[str] = None,
) -> dict:
"""
Routes the task to the optimal model and executes it.
Args:
user_message: The user's input message.
task_profile: The task profile for routing decisions.
system_prompt: Optional system prompt for the model.
Returns:
A dictionary containing the response and routing metadata.
"""
selected_tier = self._select_model_tier(task_profile)
config = MODEL_CONFIG[selected_tier]
print(f"\n[Router] Task: '{task_profile.task_description}'")
print(f"[Router] Selected: {config['description']}")
print(
f"[Router] Reason: sensitive={task_profile.contains_sensitive_data}, "
f"complexity={task_profile.complexity_score}/10, "
f"low_latency={task_profile.requires_low_latency}"
)
messages = [{"role": "user", "content": user_message}]
if config["provider"] == "ollama":
response_text = self._call_local_model(
config["model"], messages, system_prompt
)
else:
response_text = self._call_remote_model(
config["model"], messages, system_prompt
)
return {
"response": response_text,
"model_used": config["model"],
"provider": config["provider"],
"tier": selected_tier.value,
"task_profile": task_profile,
}
if __name__ == "__main__":
router = IntelligentModelRouter()
# Example 1: Sensitive HR data — must stay local
result1 = router.route_and_execute(
user_message=(
"Summarize the performance review data for the engineering "
"department Q3 2026. The data contains employee names and "
"salary information."
),
task_profile=TaskProfile(
contains_sensitive_data=True,
complexity_score=5,
requires_low_latency=False,
task_description="HR performance review summarization",
),
)
print(f"\nResponse (truncated): {result1['response'][:200]}...")
# Example 2: Complex technical reasoning, non-sensitive
result2 = router.route_and_execute(
user_message=(
"Design a fault-tolerant architecture for an industrial IoT "
"system that must maintain 99.999% uptime. Consider edge "
"computing, redundancy patterns, and graceful degradation."
),
task_profile=TaskProfile(
contains_sensitive_data=False,
complexity_score=9,
requires_low_latency=False,
task_description="Complex IoT architecture design",
),
)
print(f"\nResponse (truncated): {result2['response'][:200]}...")
# Example 3: Simple, fast classification task
result3 = router.route_and_execute(
user_message="Is this email subject line professional? 'hey u free tmrw?'",
task_profile=TaskProfile(
contains_sensitive_data=False,
complexity_score=2,
requires_low_latency=True,
task_description="Email subject line classification",
),
)
print(f"\nResponse: {result3['response']}")
The model router is a microcosm of the entire philosophy behind next-generation agentic AI. It is not about using the most powerful model for everything. It is about using the right model for each specific situation, balancing capability, privacy, latency, and cost in a way that is invisible to the user but deeply intentional in the engineering.
This kind of intelligent routing is what makes agentic systems practical at scale. A team running thousands of queries per day cannot afford to send every single one to the most expensive frontier model. But it also cannot afford to route complex reasoning tasks to a model that will produce shallow or incorrect results. The router makes these tradeoffs automatically, based on the characteristics of each task.
Chapter 9: The Human in the Loop — Trust, Oversight, and Collaboration
All of this capability raises a question that is as important as any of the technical ones: where does the human fit in? If agents can research, analyze, write, code, test, and orchestrate other agents, what is left for the person?
The answer, it turns out, is the most important things of all. Humans provide the goals. Humans provide the values. Humans provide the judgment about whether a result is actually good, not just technically correct. Humans provide the accountability that makes it safe to deploy these systems in consequential contexts.
The field has developed a concept called "human-in-the-loop" (HITL) design, which refers to architectural patterns that keep humans appropriately involved in agentic workflows. The key word is "appropriately." The goal is not to have humans approve every single action an agent takes — that would eliminate most of the efficiency gains. The goal is to identify the decision points where human judgment is genuinely necessary and to design the system to pause and seek that judgment at exactly those moments.
In practice, this means defining "checkpoints" in the workflow where the agent presents its intermediate results and asks for confirmation before proceeding. It means building "guardrails" that prevent agents from taking certain classes of actions without explicit human approval. It means designing "audit trails" that record every action an agent takes so that humans can review and understand what happened after the fact.
LangGraph has first-class support for human-in-the-loop patterns through its interrupt() mechanism, which allows a workflow to pause at any node and wait for human input before continuing. The workflow state is persisted by a checkpointer (here MemorySaver) so that no work is lost during the pause. Resumption is handled by passing a Command(resume=value) object to the next invoke() call.
Install dependencies:
pip install langgraph>=1.2.11 langchain-openai>=1.5.0
Run:
python human_in_the_loop.py
"""
human_in_the_loop.py
Demonstrates the human-in-the-loop pattern using LangGraph's interrupt()
mechanism. The workflow pauses at a review checkpoint and waits for
explicit human approval before executing consequential actions.
Resumption uses Command(resume=value) as required by LangGraph v1.x.
This pattern is essential for maintaining trust and oversight in agentic
systems that operate in high-stakes or regulated environments.
Tested with: langgraph>=1.2.11, langchain-openai>=1.5.0, Python 3.10+
"""
import os
from typing import Annotated, TypedDict
from langchain_core.messages import AIMessage, BaseMessage, HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages
from langgraph.types import Command, interrupt
class ContentWorkflowState(TypedDict):
"""
State for the content generation and approval workflow.
Attributes:
messages: Full conversation and agent output history.
draft_content: The generated draft awaiting human review.
human_approved: Whether the human has approved the draft.
final_content: The published content after approval.
revision_notes: Human feedback for revision if not approved.
"""
messages: Annotated[list[BaseMessage], add_messages]
draft_content: str
human_approved: bool
final_content: str
revision_notes: str
def content_generation_node(state: ContentWorkflowState) -> dict:
"""
Generates a draft based on the user's request.
This node produces content that will be reviewed by a human before
any further processing or publication occurs.
Args:
state: The current workflow state.
Returns:
Partial state update with the generated draft.
"""
llm = ChatOpenAI(
model="gpt-5",
temperature=0.6,
api_key=os.environ.get("OPENAI_API_KEY"),
)
user_request = state["messages"][0].content
generation_prompt = (
f"Generate a professional internal communication based on this "
f"request: {user_request}\n\n"
"The communication should be clear, concise, and appropriate for a "
"corporate environment. Include a subject line, greeting, body, and "
"professional closing."
)
response = llm.invoke([HumanMessage(content=generation_prompt)])
draft = response.content
return {
"messages": [AIMessage(content=f"Draft generated:\n\n{draft}")],
"draft_content": draft,
"human_approved": False,
"final_content": "",
"revision_notes": "",
}
def human_review_node(state: ContentWorkflowState) -> dict:
"""
Pauses the workflow and waits for human review and approval.
The interrupt() call suspends graph execution and surfaces the draft
to the calling application. The graph resumes only when the caller
invokes it again with Command(resume=<human_decision>).
Args:
state: The current workflow state containing the draft.
Returns:
Partial state update with the human's approval decision.
"""
draft = state["draft_content"]
# interrupt() suspends execution. The dict passed here is available
# to the caller via the interrupt value in the graph's stream/invoke
# response, so the UI can display it to the human reviewer.
human_decision: str = interrupt(
{
"action": "review_required",
"draft": draft,
"instructions": (
"Please review the draft above. Respond with:\n"
" 'approve' — to publish as-is\n"
" 'revise: <notes>' — to request specific changes"
),
}
)
decision_text = human_decision.strip().lower()
if decision_text == "approve":
return {
"messages": [HumanMessage(content="Human reviewer: APPROVED")],
"human_approved": True,
"revision_notes": "",
}
if decision_text.startswith("revise:"):
revision_notes = human_decision[7:].strip()
return {
"messages": [
HumanMessage(
content=f"Human reviewer: REVISION REQUESTED — {revision_notes}"
)
],
"human_approved": False,
"revision_notes": revision_notes,
}
# Unrecognised response — default to requesting a revision
return {
"messages": [
HumanMessage(
content="Human reviewer: unclear response, defaulting to revision"
)
],
"human_approved": False,
"revision_notes": "Please clarify the intent and tone.",
}
def revision_node(state: ContentWorkflowState) -> dict:
"""
Revises the draft based on human feedback.
Receives the reviewer's notes and produces an improved version of the
draft, which then loops back through the review checkpoint.
Args:
state: The current workflow state with revision notes.
Returns:
Partial state update with the revised draft.
"""
llm = ChatOpenAI(
model="gpt-5",
temperature=0.5,
api_key=os.environ.get("OPENAI_API_KEY"),
)
revision_prompt = (
f"Original draft:\n{state['draft_content']}\n\n"
f"Reviewer's feedback:\n{state['revision_notes']}\n\n"
"Please revise the draft to address the feedback. Maintain the "
"professional tone and corporate communication style."
)
response = llm.invoke([HumanMessage(content=revision_prompt)])
revised_draft = response.content
return {
"messages": [AIMessage(content=f"Revised draft:\n\n{revised_draft}")],
"draft_content": revised_draft,
"human_approved": False,
}
def publication_node(state: ContentWorkflowState) -> dict:
"""
Publishes the approved content.
This node only executes after explicit human approval. In a real
system, this would send the email, post to the intranet, or trigger
whatever publication mechanism is appropriate.
Args:
state: The current workflow state with the approved draft.
Returns:
Partial state update confirming publication.
"""
approved_content = state["draft_content"]
# In production: send email, post to CMS, notify stakeholders, etc.
publication_confirmation = (
"Content successfully published.\n"
"Publication timestamp: 2026-08-14T12:15:00Z\n"
f"Content preview: {approved_content[:100]}..."
)
return {
"messages": [AIMessage(content=publication_confirmation)],
"final_content": approved_content,
}
def route_after_review(state: ContentWorkflowState) -> str:
"""
Conditional edge: routes to publication if approved, revision if not.
Args:
state: The current workflow state.
Returns:
'publish' if approved, 'revise' if revision is needed.
"""
return "publish" if state["human_approved"] else "revise"
def build_approval_workflow() -> StateGraph:
"""
Constructs the human-in-the-loop content approval workflow graph.
Graph structure:
START -> generate -> human_review -> [publish | revise -> human_review]
The revision loop allows multiple rounds of feedback until the human
is satisfied with the content.
The MemorySaver checkpointer persists state across interrupt/resume
cycles so that no work is lost while waiting for human input.
Returns:
A compiled StateGraph with memory checkpointing enabled.
"""
graph = StateGraph(ContentWorkflowState)
graph.add_node("generate", content_generation_node)
graph.add_node("human_review", human_review_node)
graph.add_node("revise", revision_node)
graph.add_node("publish", publication_node)
graph.add_edge(START, "generate")
graph.add_edge("generate", "human_review")
graph.add_conditional_edges(
"human_review",
route_after_review,
{"publish": "publish", "revise": "revise"},
)
graph.add_edge("revise", "human_review") # Loop back for re-review
graph.add_edge("publish", END)
# MemorySaver persists state between the initial invoke() and the
# resume invoke() that follows the human's decision.
memory = MemorySaver()
return graph.compile(checkpointer=memory)
if __name__ == "__main__":
workflow = build_approval_workflow()
# Each workflow run needs a unique thread_id for state persistence
config = {"configurable": {"thread_id": "content-workflow-001"}}
initial_state: ContentWorkflowState = {
"messages": [
HumanMessage(
content=(
"Write an internal announcement about the new flexible "
"working hours policy starting September 2026. Employees "
"can now choose their core hours between 9 am and 3 pm."
)
)
],
"draft_content": "",
"human_approved": False,
"final_content": "",
"revision_notes": "",
}
# Run until the first interrupt (human review checkpoint)
print("Running workflow until human review checkpoint...")
result = workflow.invoke(initial_state, config=config)
print("\nWorkflow paused for human review.")
print("Draft content:")
print("-" * 50)
print(result.get("draft_content", "No draft generated"))
print("-" * 50)
# Simulate human approval.
# In a real system this value arrives from a web UI, Slack bot, email
# reply parser, or any other human-facing interface.
human_input = "approve"
print(f"\nHuman decision: {human_input}")
# Resume the workflow by passing Command(resume=<value>).
# This is the correct LangGraph v1.x API for resuming after interrupt().
final_result = workflow.invoke(
Command(resume=human_input),
config=config,
)
print("\nWorkflow completed.")
print(f"Final content published: {bool(final_result.get('final_content'))}")
The human-in-the-loop pattern is not just a safety mechanism. It is a collaboration model. The agent does the heavy lifting of generation and revision. The human provides the judgment and approval that gives the output its legitimacy. Neither can do the job as well alone as they can together.
This is, perhaps, the most important insight in all of agentic AI. The goal is not to replace human judgment. The goal is to amplify human capability by offloading the mechanical, repetitive, and time-consuming parts of knowledge work to agents, while keeping humans in the loop for the decisions that genuinely require human wisdom, accountability, and values.
Chapter 10: The Road Ahead — What Comes Next
We are, by any reasonable measure, still in the early days of agentic AI. The systems we have described in this article are impressive, but they are also fragile in ways that matter. Agents still hallucinate. They still get confused by ambiguous instructions. They still fail in unexpected ways when they encounter situations that fall outside their training distribution. Orchestration systems still require careful engineering to handle edge cases and failure modes. Memory systems still struggle with the problem of what to remember and what to forget.
But the trajectory is clear, and it is steep.
One of the most interesting things happening right now is the emergence of genuinely persistent agent identities. Today, even the best memory systems give agents a kind of episodic continuity — the ability to remember specific past interactions. What is emerging is something closer to genuine accumulated expertise: agents that get better at working with a specific person or on a specific domain over time, not just because they remember past interactions but because they have developed refined internal representations of that person's preferences, working style, and domain knowledge. That is a qualitatively different kind of memory, and it is closer to how human expertise actually develops.
Closely related is the standardization of agent communication protocols. Today, multi-agent systems are largely proprietary — a CrewAI agent cannot easily talk to a LangGraph agent, which cannot easily talk to an AutoGen agent. This is changing rapidly. The industry is converging on open standards for agent communication, including Anthropic's Model Context Protocol (MCP), which has gained broad adoption through 2025 and 2026. When agents from different vendors and frameworks can collaborate seamlessly, the kinds of systems you can build become dramatically more interesting.
Something that often gets less attention but matters enormously is the integration of agents with physical systems. The digital-physical boundary is dissolving. Agents that can control industrial equipment, navigate physical spaces, and interact with the physical world through robotic systems are moving from research labs into production environments. This convergence — software intelligence meeting physical infrastructure — is arguably the most consequential development of the next decade.
Then there is the question of agents that can genuinely learn from their mistakes. Today, when an agent fails at a task, that failure is typically lost. The agent does not update its behavior based on the failure. Research into online learning, reinforcement learning from human feedback, and constitutional AI is beginning to change this, creating agents that genuinely improve through experience rather than just through retraining.
And underneath all of it runs a deeper question that the field is actively grappling with: what happens when agents become capable enough to improve themselves? This is the territory of recursive self-improvement, and it is where the conversation about AI safety becomes most urgent and most important. The field is actively working on alignment techniques, interpretability tools, and governance frameworks to ensure that increasingly capable agents remain aligned with human values and subject to human oversight. These are not abstract philosophical concerns — they are engineering problems that need engineering solutions, and some of the best minds in the field are working on them right now.
The pace of progress is such that the system you build today will look quaint compared to what will be possible next year. That is both thrilling and sobering. The right response is not to be paralyzed by the pace of change, but to build thoughtfully, to keep humans appropriately in the loop, to prioritize safety and alignment alongside capability, and to approach the technology with the same combination of curiosity, rigor, and ethical seriousness that the best engineers bring to every hard problem.
Chapter 11: Practical Guidance for Getting Started Today
If you have read this far and you are wondering how to actually start building with these technologies, here is a practical roadmap organized by experience level and ambition.
If you are new to agentic AI, the best starting point is a single-agent system with tool calling. Pick a task you do repeatedly that involves looking up information and synthesizing it into a response. Build a simple agent with two or three tools that automate that task. Use the OpenAI API or Ollama, depending on your data sensitivity requirements. Get comfortable with the tool-calling loop before adding any complexity. The agent_tool_calling.py example in Chapter 2 is your starting point.
Once you are comfortable with single-agent tool calling, add a memory system. Start with ChromaDB running locally. Build the store-and-retrieve pattern shown in Chapter 4. Run the agent over several sessions and observe how it begins to feel more personalized and contextually aware. This is the moment when the technology stops feeling like a demo and starts feeling like a genuine productivity tool.
When you are ready for multi-agent systems, start with CrewAI's sequential process. It is the simplest orchestration pattern and the easiest to debug. Build a three-agent pipeline for a workflow you know well — something like research, analysis, and reporting. Observe how the specialization improves the quality of each stage's output compared to asking a single agent to do everything.
When you need adaptive orchestration, move to LangGraph. Start with a simple two-node graph and add complexity incrementally. Add the human-in-the-loop pattern early, before you deploy anything consequential. It is much easier to add oversight mechanisms at design time than to retrofit them later.
Throughout all of this, keep a few principles close at hand. Privacy and data governance must be designed in from the start, not added as an afterthought — use local models for sensitive data. Audit trails are not optional in regulated environments. Test your agents adversarially, not just with happy-path inputs. And always keep a human in the loop for decisions that genuinely matter.
The technology is ready. The frameworks are mature. The models are capable. The question is not whether to build with agentic AI, but how to build well.
Conclusion: The Partnership Model
We began this article by talking about the moment when technological change stops being theoretical and starts being personal. That moment is here, and it is characterized by a shift in the fundamental relationship between humans and AI systems.
The old model was transactional. You typed a question, the model returned an answer, and the interaction was over. The new model is collaborative. You work alongside agents that remember you, understand your context, coordinate with specialized colleagues, and adapt their behavior based on what they learn about you and your work. The agent is not a search engine. It is not a calculator. It is a cognitive partner.
This shift brings both tremendous opportunity and genuine responsibility. The opportunity is to amplify human capability in ways that were previously impossible — to free knowledge workers from the mechanical and repetitive parts of their work so they can focus on the creative, strategic, and deeply human parts. The responsibility is to build these systems thoughtfully, to maintain appropriate human oversight, to be honest about their limitations, and to ensure that the benefits are distributed broadly rather than concentrated narrowly.
The architecture we have explored in this article — tool-calling agents, long-term memory systems, massive context windows, specialized agent teams, and human-in-the-loop oversight — is not just a technical design. It is a vision of what the collaboration between humans and AI can look like when it is done well. It is a vision worth working toward.
The agents are ready. The question is what we will build with them.
No comments:
Post a Comment