DEV Community: Alexsandro Souza

Building a Production-Ready AI Agent Harness

Alexsandro Souza — Mon, 16 Mar 2026 15:50:39 +0000

Building an AI agent is 10% "The Brain" and 90% "The Plumbing." AI can write your agent in an afternoon. What it can't do is tell you what production looks like.

It won't ask about authentication, rate limiting, or what happens when your LLM hallucinates a credit card number into a response. It won't add guardrails that block prompt injection before the message reaches the model, or redact PII from the output before it reaches the user. It won't wire up tracing so you can replay exactly what the agent did six calls deep, persist conversation state across restarts, or set up the metrics dashboard that tells you why latency spiked at 3 AM. It won't build the eval pipeline that catches quality regressions before your users do, handle context overflow when a tool dumps 80,000 characters into the conversation, or add retry logic with exponential backoff so a single upstream timeout doesn't crash the entire session. That part is still on you.

And yet, most LangGraph and LangChain tutorials stop right before any of this matters -- a working agent in a Jupyter notebook that answers questions, uses tools, and streams tokens. The gap between that demo and a service you'd actually deploy is enormous, and every team fills it by reinventing the same infrastructure from scratch.

An agent harness is the answer to this problem. Think of it like a test harness, but for production concerns instead of assertions. It is the infrastructure shell that wraps around your agent -- handling authentication, guardrails, memory, state persistence, observability, and deployment -- so the agent itself only contains the logic that makes it unique. You write the brain; the harness provides the skeleton, the nervous system, and the immune system.

This matters because AI agents are not stateless functions. They hold conversations across sessions, remember user preferences, call unpredictable external tools, and produce outputs that need safety checks before reaching the user. Without a harness, every one of these concerns leaks into your agent code, turning a clean graph into a tangled mess of infrastructure that is impossible to reuse across agents.

This article walks through the architecture of such a harness -- built with LangGraph, FastAPI, Langfuse, PostgreSQL with pgvector, MCP and AI Skills. The full implementation is open source. You define the agent logic, the harness provides everything else.

1. Project Architecture

The codebase separates agent logic from infrastructure through a clean directory structure:

src/
├── app/
│   ├── agents/              # Self-contained agent directories
│   │   ├── chatbot/         # Reference agent implementation
│   │   ├── open_deep_research/  # Multi-agent research workflow
│   │   ├── text_to_sql/     # Text-to-SQL agent
│   │   └── tools/           # Shared tools (search, think)
│   ├── api/
│   │   ├── v1/              # Versioned API routes
│   │   ├── metrics/         # Prometheus middleware
│   │   ├── security/        # Auth and rate limiting
│   │   └── logging_context.py
│   └── core/                # Shared infrastructure
│       ├── guardrails/      # Input/output safety
│       ├── context/         # LLM context overflow prevention
│       ├── memory/          # Long-term memory (mem0)
│       ├── checkpoint/      # State persistence
│       ├── mcp/             # Model Context Protocol
│       ├── metrics/         # Prometheus definitions
│       ├── llm/             # LLM utilities
│       ├── db/              # Database connections
│       └── common/          # Config, logging, models
├── evals/                   # Evaluation framework
├── mcp/                     # Sample MCP server
└── cli/                     # CLI clients

The key design decision: agents are self-contained directories under src/app/agents/. Each agent defines its own graph, system prompt, and tools. Everything else -- auth, memory, checkpointing, guardrails, metrics -- is injected by the harness through src/app/core/.

The harness ships with three reference agents, each demonstrating a different architecture approach:

Agent	Architecture	Pattern
Chatbot	Custom graph workflow	Linear pipeline with tool loop and guardrail nodes
Deep Research	Multi-agent supervisor/researcher	Nested subgraphs with parallel delegation
Text-to-SQL	ReAct (via Deep Agents)	Autonomous reasoning-action loop with skills and memory files

2. Agent Architecture Approaches

The harness does not prescribe a single agent pattern. Different tasks call for different architectures. The three included agents demonstrate the spectrum from simple to complex.

Approach 1: Custom Graph Workflow (Chatbot)

The chatbot agent uses a hand-crafted LangGraph StateGraph where every node and edge is explicitly defined. This gives full control over the execution flow.

The graph is built in _create_graph():

async def _create_graph(self) -> StateGraph:
    input_guardrail = create_input_guardrail_node(next_node="chat")
    output_guardrail = create_output_guardrail_node()

    graph_builder = StateGraph(GraphState)
    graph_builder.add_node("input_guardrail", input_guardrail, ends=["chat", END])
    graph_builder.add_node("chat", self._chat_node, ends=["tool_call", "output_guardrail"])
    graph_builder.add_node("tool_call", self._tool_call_node, ends=["chat"])
    graph_builder.add_node("output_guardrail", output_guardrail)
    graph_builder.set_entry_point("input_guardrail")
    graph_builder.add_edge("output_guardrail", END)
    return graph_builder

Each node returns a Command that controls routing. The chat node decides whether to route to tools or the output guardrail based on the LLM response:

async def _chat_node(self, state: GraphState, config: RunnableConfig) -> Command:
    system_prompt = load_system_prompt(long_term_memory=state.long_term_memory)
    messages = prepare_messages(state.messages, chatbot_model, system_prompt)

    model = chatbot_model.bind_tools(self._get_all_tools()).with_retry(stop_after_attempt=3)

    response_message = await model_invoke_with_metrics(
        model, dump_messages(messages), settings.DEFAULT_LLM_MODEL, self.name, config
    )
    response_message = process_llm_response(response_message)

    goto = "tool_call" if response_message.tool_calls else "output_guardrail"
    return Command(update={"messages": [response_message]}, goto=goto)

The state schema is minimal -- LangGraph's add_messages reducer handles message accumulation:

class GraphState(MessagesState):
    long_term_memory: str

System prompts are markdown templates with placeholders that get filled at invocation time:

# Name: {agent_name}
# Role: A friendly and professional assistant

## What you know about the user
{long_term_memory}

## Current date and time
{current_date_and_time}

When to use this approach: General-purpose assistants, chatbots, and any agent where you want explicit control over every routing decision. The graph is easy to visualize, debug, and extend with new nodes.

Approach 2: Multi-Agent Supervisor/Researcher (Deep Research)

The deep research agent uses nested subgraphs -- a supervisor that plans research, then delegates to multiple researcher subagents running in parallel. Each subagent is itself a compiled LangGraph with its own ReAct-style tool loop.

The main graph composes these subgraphs as nodes:

def _build_deep_research_graph(self) -> StateGraph:
    deep_researcher_builder = StateGraph(AgentState, input=AgentInputState)

    deep_researcher_builder.add_node("input_guardrail", input_guardrail, ends=["clarify_with_user", END])
    deep_researcher_builder.add_node("clarify_with_user", clarify_with_user)
    deep_researcher_builder.add_node("write_research_brief", write_research_brief)
    deep_researcher_builder.add_node("research_supervisor", self.supervisor_subagent.get_graph())
    deep_researcher_builder.add_node("final_report_generation", final_report_generation)
    deep_researcher_builder.add_node("output_guardrail", output_guardrail)

    deep_researcher_builder.add_edge("research_supervisor", "final_report_generation")
    deep_researcher_builder.add_edge("final_report_generation", "output_guardrail")
    deep_researcher_builder.add_edge("output_guardrail", END)
    return deep_researcher_builder

The supervisor delegates research tasks to sub-researchers in parallel using asyncio.gather:

research_tasks = [
    self._researcher_agent.agent_invoke({
        "researcher_messages": [HumanMessage(content=tool_call["args"]["research_topic"])],
        "research_topic": tool_call["args"]["research_topic"]
    }, "session_id_placeholder", 1)
    for tool_call in allowed_conduct_research_calls
]
tool_results = await asyncio.gather(*research_tasks)

Each researcher runs its own ReAct loop -- calling search tools, reflecting with think_tool, and iterating up to MAX_REACT_TOOL_CALLS before compressing findings into a summary.

When to use this approach: Complex research tasks, multi-step analysis, or any workflow that benefits from decomposition into parallel sub-tasks. The subgraph composition lets you scale horizontally by adding more specialized agents.

Approach 3: ReAct with Deep Agents (Text-to-SQL)

The text-to-SQL agent takes a fundamentally different approach: instead of hand-crafting a graph, it uses the Deep Agents framework (deepagents library) to create an autonomous ReAct agent with persistent skills and memory files.

def create_sql_deep_agent():
    db = SQLDatabase.from_uri(f"sqlite:///{db_path}", sample_rows_in_table_info=3)
    model = ChatOpenAI(model="gpt-5-mini", reasoning={"effort": "medium"}, temperature=0)

    toolkit = SQLDatabaseToolkit(db=db, llm=model)
    sql_tools = toolkit.get_tools()

    agent = create_deep_agent(
        model=model,
        memory=["./AGENTS.md"],
        skills=["./skills/"],
        tools=sql_tools,
        subagents=[],
        backend=FilesystemBackend(root_dir=base_dir),
    )
    return agent

The agent is configured through markdown files rather than code:

AGENTS.md -- Agent identity, role definition, safety rules (read-only SQL access), and planning strategies
skills/ -- Specialized workflows defined as markdown skill files (schema-exploration, query-writing)
FilesystemBackend -- Persistent storage for intermediate results and planning artifacts

Skills: Extending Agents with Natural Language

Skills are emerging as a modern pattern for extending agent capabilities -- and for good reason. Instead of writing new Python functions or graph nodes, you define a skill as a markdown file that describes when to activate and how to execute a workflow. The agent reads skill descriptions in its context and loads the full instructions on demand, only when the task requires it.

The text-to-SQL agent ships with two skills. The schema-exploration skill teaches the agent how to discover database structure:

---
name: schema-exploration
description: For discovering and understanding database structure, tables, columns, and relationships
---

## When to Use This Skill
Use this skill when you need to:
- Understand the database structure
- Find which tables contain certain types of data
- Discover column names and data types

## Workflow
### 1. List All Tables
Use `sql_db_list_tables` tool to see all available tables in the database.

### 2. Get Schema for Specific Tables
Use `sql_db_schema` tool with table names to examine columns, data types, and relationships.

### 3. Map Relationships
Identify how tables connect via foreign keys.

The query-writing skill teaches the agent a structured approach to SQL generation -- planning with write_todos for complex queries, validating JOIN conditions, and enforcing safety rules like LIMIT clauses and read-only access.

This is progressive disclosure applied to agent design. The agent always sees the skill names and descriptions, but it only loads the full multi-page instructions when it decides a skill is relevant. This keeps the context window lean for simple questions while providing deep, step-by-step expertise for complex ones. Adding a new capability is as simple as dropping a new .md file into the skills/ directory -- no code changes, no redeployment, no graph rewiring.

The wrapper class applies the same harness guardrails (content filter, PII blocking, output safety check) before and after the ReAct loop:

class TextSQLDeepAgent:
    async def agent_invoke(self, agent_input, session_id, user_id=None):
        query = agent_input.get("query", "")

        filter_result = check_content_filter(query)
        if filter_result.is_blocked:
            return [Message(role="assistant", content="I cannot process this request.")]

        pii_findings = detect_pii(query, pii_types=[PIIType.API_KEY, PIIType.SSN, PIIType.CREDIT_CARD])
        if pii_findings:
            return [Message(role="assistant", content="Your message contains sensitive information.")]

        response = await self.agent.ainvoke({"messages": [{"role": "user", "content": query}]}, config=config)
        messages = process_messages(response["messages"])
        await self.process_safe_output(messages)
        return messages

When to use this approach: Domain-specific agents where the reasoning strategy is better expressed in natural language than graph code. The ReAct pattern lets the agent decide its own execution order -- explore schema, write queries, fix errors, and iterate -- without you encoding every possible path.

Architecture Comparison

The three approaches represent different trade-offs:

Dimension	Custom Graph	Multi-Agent Subgraphs	ReAct (Deep Agents)
Control	Full -- every edge is explicit	Hierarchical -- control delegation boundaries	Minimal -- agent decides its own path
Debuggability	High -- deterministic flow	Medium -- subgraph boundaries are clear	Lower -- emergent behavior
Parallelism	Manual	Built-in via supervisor delegation	Sequential by default
Flexibility	Add/remove nodes in code	Compose new subgraph combinations	Change behavior via markdown files
Best for	Chat, simple workflows	Research, analysis, decomposable tasks	Domain-specific tool use (SQL, APIs)

All three share the same harness infrastructure: guardrails, Langfuse tracing, Prometheus metrics, structured logging, and authentication. The architecture choice only affects what lives inside src/app/agents/<your_agent>/.

3. Guardrails: Input and Output Safety

The harness provides guardrails as factory functions that return LangGraph-compatible node functions. This lets each agent customize its safety checks without reimplementing the pattern.

Input Guardrails

Input guardrails run deterministic checks before the message reaches the LLM:

def create_input_guardrail_node(
    next_node: str,
    banned_keywords: list[str] | None = None,
    pii_check_enabled: bool = True,
    prompt_injection_check: bool = True,
    block_pii_types: list[PIIType] | None = None,
) -> Callable:

Two layers of validation run in sequence:

1. Content filter -- Banned keyword matching and prompt injection detection using regex patterns:

PROMPT_INJECTION_PATTERNS: list[str] = [
    r"ignore\s+(all\s+)?(previous|prior|above)\s+(instructions|prompts|rules)",
    r"disregard\s+(all\s+)?(previous|prior|above)\s+(instructions|prompts|rules)",
    r"you\s+are\s+now\s+(?:a|an|in)\s+(?:jailbreak|unrestricted|evil|DAN)",
    r"override\s+(?:your|all|system)\s+(?:instructions|rules|constraints)",
    r"reveal\s+(?:your|the|system)\s+(?:prompt|instructions|rules)",
]

2. PII detection -- Regex-based scanning for sensitive data types (SSN, API keys, credit cards) with Luhn validation for credit card numbers:

class PIIType(str, Enum):
    EMAIL = "email"
    CREDIT_CARD = "credit_card"
    IP = "ip"
    API_KEY = "api_key"
    PHONE = "phone"
    SSN = "ssn"

When input is blocked, the guardrail routes directly to END with a rejection message:

if filter_result.is_blocked:
    return Command(
        update={"messages": [AIMessage(content=BLOCKED_INPUT_MESSAGE)]},
        goto=END,
    )

Output Guardrails

Output guardrails apply two checks to the agent's response before it reaches the user:

1. Deterministic PII redaction -- Scans the output and replaces detected PII with [REDACTED_EMAIL], [REDACTED_SSN], etc. Supports multiple strategies: redact, mask, hash, or block.

2. LLM-based safety evaluation -- Uses a fast, low-cost model to classify the response as SAFE or UNSAFE. Unsafe responses are replaced with a standard message:

async def evaluate_safety(content: str) -> bool:
    model = _get_safety_model()
    prompt = SAFETY_EVALUATION_PROMPT.format(response=content[:2000])
    result = await model.ainvoke([{"role": "user", "content": prompt}])
    verdict = result.content.strip().upper()
    return "UNSAFE" not in verdict

The output guardrail is fail-open: if the safety check itself errors, the response passes through. This prevents the safety system from becoming a point of failure.

4. Long-Term Memory: Semantic Memory with mem0 and pgvector

The harness provides per-user semantic memory backed by pgvector. Before every agent invocation, relevant memories are retrieved by semantic similarity. After invocation, new memories are extracted from the conversation.

async def get_relevant_memory(user_id: int, query: str) -> str:
    memory = await get_memory_instance()
    results = await memory.search(user_id=str(user_id), query=query)
    return "\n".join([f"* {result['memory']}" for result in results["results"]])

Memory updates happen in the background to avoid blocking the response:

def bg_update_memory(user_id: int, messages: list[dict], metadata: dict = None) -> None:
    asyncio.create_task(update_memory(user_id, messages, metadata))

The agent orchestrates this in its agent_invoke method -- retrieve first, update after:

async def agent_invoke(self, messages, session_id, user_id=None):
    relevant_memory = (await get_relevant_memory(user_id, messages[-1].content)) or "No relevant memory found."
    agent_input = {"messages": dump_messages(messages), "long_term_memory": relevant_memory}

    response = await self._graph.ainvoke(input=agent_input, config=config)
    # ... process response ...

    bg_update_memory(user_id, messages_dic, {"session_id": session_id, "agent_name": self.name})
    return result

The memory singleton connects to pgvector using the same PostgreSQL instance as the rest of the application, configured through the shared Settings class.

5. Context Management: Preventing LLM Context Overflow

Long-running conversations and large tool results can push the message history past the model's context window. The harness addresses this with a two-layer strategy in src/app/core/context/: evict oversized tool results immediately, and summarize the conversation history before it overflows.

Layer 1: Tool Result Eviction

When a tool returns a result exceeding ~80,000 characters (~20K tokens), the full output is written to disk and the in-state message is replaced with a compact head/tail preview:

MAX_TOOL_RESULT_CHARS = 80_000
PREVIEW_LINES = 5
MAX_LINE_CHARS = 1_000

def truncate_tool_call_if_too_long(tool_message: ToolMessage) -> ToolMessage:
    content = tool_message.content
    if not isinstance(content, str) or len(content) <= MAX_TOOL_RESULT_CHARS:
        return tool_message

    file_path = _write_large_result(tool_message.tool_call_id, content)
    preview = _build_preview(content)

    evicted_content = (
        f"[Tool result too large ({len(content):,} chars). "
        f"Full output saved to: {file_path}]\n\n"
        f"--- Preview (first {PREVIEW_LINES} lines) ---\n"
        f"{preview['head']}\n...\n"
        f"--- Preview (last {PREVIEW_LINES} lines) ---\n"
        f"{preview['tail']}\n\n"
        f'Use read_file("{file_path}") to access the full content.'
    )

    return ToolMessage(content=evicted_content, name=tool_message.name, tool_call_id=tool_message.tool_call_id)

The preview gives the LLM enough context to decide whether it needs the full output. If it does, it can call read_file on the saved path to retrieve the content in chunks. This wrapping is applied at every tool-result creation site -- built-in tools, MCP tools, and parallel tool execution:

# In the tool_call node
outputs.append(truncate_tool_call_if_too_long(
    ToolMessage(content=tool_result, name=tool_name, tool_call_id=tool_call["id"])
))

Layer 2: Conversation Summarization

Even with tool results evicted, conversations accumulate context over many turns. The summarize_if_too_long function runs before every LLM call and triggers when the conversation reaches 85% of the model's context window:

async def _chat_node(self, state: GraphState, config: RunnableConfig) -> Command:
    condensed_messages = await summarize_if_too_long(
        messages=state.messages,
        model_name=f"openai:{settings.DEFAULT_LLM_MODEL}",
        llm=chatbot_model,
        session_id=config["configurable"]["thread_id"],
    )

    system_prompt = load_system_prompt(long_term_memory=state.long_term_memory)
    messages = prepare_messages(condensed_messages, chatbot_model, system_prompt)
    # ... LLM call ...

Token counting uses count_tokens_approximately from langchain-core for the threshold check. The cheaper character-based heuristic (NUM_CHARS_PER_TOKEN = 4) is used for the split-point search and argument truncation to avoid repeated full counts.

The function is a no-op when the context is within budget. When it triggers, it proceeds in two stages:

Stage 1 -- Lightweight truncation (no LLM call). Tool-call arguments in older messages -- like file contents passed to write_file -- are truncated to 2K characters. If this alone brings the count below the threshold, no summarization is needed:

def _truncate_tool_call_args(messages: list[BaseMessage]) -> list[BaseMessage]:
    result = []
    for msg in messages:
        if isinstance(msg, AIMessage) and msg.tool_calls:
            truncated_calls = []
            for tc in msg.tool_calls:
                new_args = {}
                for key, value in tc.get("args", {}).items():
                    if isinstance(value, str) and len(value) > TOOL_ARG_TRUNCATE_CHARS:
                        new_args[key] = value[:TOOL_ARG_TRUNCATE_CHARS] + f"\n... [truncated, {len(value):,} chars total]"
                    else:
                        new_args[key] = value
                truncated_calls.append({**tc, "args": new_args})
            result.append(msg.model_copy(update={"tool_calls": truncated_calls}))
        else:
            result.append(msg)
    return result

Stage 2 -- Full summarization. If truncation was not enough, the conversation is split into "old" and "recent" segments. The split snaps to a HumanMessage boundary so tool-call chains are never broken. The old segment is written in full to a markdown file, and an LLM generates a concise summary that replaces it:

file_path = _write_messages_to_markdown(old_messages, session_id)
summary_text = await _generate_summary(truncated_old, llm)

summary_message = HumanMessage(
    content=(
        f"[Summary of earlier conversation ({len(old_messages)} messages). "
        f"Full transcript: {file_path}]\n\n{summary_text}"
    )
)

return [summary_message] + recent_messages

The split-point algorithm walks backward from the end of the message list, accumulating characters until it fills ~30% of the model's context window for the "recent" portion. It then snaps forward to the next HumanMessage to find a clean boundary:

def _find_safe_split_index(messages, token_limit):
    recent_chars_budget = int(token_limit * RECENT_CONTEXT_RATIO * NUM_CHARS_PER_TOKEN)
    chars_from_end = 0

    for i in range(len(messages) - 1, -1, -1):
        chars_from_end += _estimate_message_chars(messages[i])
        if chars_from_end >= recent_chars_budget:
            raw_split = i + 1
            break

    # Snap to nearest HumanMessage boundary
    for j in range(raw_split, len(messages)):
        if isinstance(messages[j], HumanMessage):
            return j
    return 0

If the summarization LLM call itself fails, the function falls back to a plain message-role listing so the conversation can continue without crashing. The full old messages are always preserved in the markdown file regardless of whether the LLM summary succeeds.

6. State Persistence: Conversation Checkpointing

LangGraph's AsyncPostgresSaver automatically persists the full graph state after every node execution. This means conversations survive server restarts and can be resumed from exactly where they left off.

async def get_checkpointer():
    connection_pool = await get_connection_pool()
    if connection_pool:
        checkpointer = AsyncPostgresSaver(connection_pool)
        await checkpointer.setup()
    else:
        checkpointer = None
        if settings.ENVIRONMENT != Environment.PRODUCTION:
            raise Exception("Connection pool initialization failed")
    return checkpointer

The checkpointer is compiled into the graph once and reused:

self._graph = graph_builder.compile(checkpointer=self.checkpointer, name=self.name)

Every invocation passes a thread_id (the session ID) so LangGraph knows which conversation to resume:

config["configurable"] = {"thread_id": session_id}

When a user deletes a session, the harness cleans up checkpoint tables:

async def clear_checkpoints(session_id: str) -> None:
    async with conn_pool.connection() as conn:
        for table in settings.CHECKPOINT_TABLES:
            await conn.execute(f"DELETE FROM {table} WHERE thread_id = %s", (session_id,))

7. Authentication and Session Management

The harness uses JWT-based authentication with a session-per-conversation model. Users register, log in to get a user-scoped token, then create sessions -- each session represents a separate conversation thread.

Protected endpoints use FastAPI dependency injection:

@router.post("/chat", response_model=ChatResponse)
@limiter.limit(settings.RATE_LIMIT_ENDPOINTS["chat"][0])
async def chat(
    request: Request,
    chat_request: ChatRequest,
    session: Session = Depends(get_current_session),
):

Rate limiting is configured per-endpoint through environment variables:

default_endpoints = {
    "chat": ["30 per minute"],
    "chat_stream": ["20 per minute"],
    "deep_research": ["10 per minute"],
    "register": ["10 per hour"],
    "login": ["20 per minute"],
}

All user input passes through sanitization that strips HTML, removes <script> tags, and filters null bytes -- applied at both the API and validation layers.

8. Observability: Tracing, Metrics, and Structured Logging

Production observability requires three complementary systems. The harness integrates all three.

Langfuse Tracing

Every LLM call is traced through Langfuse via a callback handler injected into the agent config:

self._config = {
    "callbacks": [langfuse_callback_handler],
    "metadata": {
        "environment": settings.ENVIRONMENT.value,
        "debug": settings.DEBUG,
    },
}

This captures the full chain of LLM calls, tool invocations, token usage, and latency for every conversation turn -- without modifying agent code.

Prometheus Metrics

Custom metrics track both infrastructure and LLM-specific performance:

llm_inference_duration_seconds = Histogram(
    "llm_inference_duration_seconds",
    "Time spent processing LLM inference",
    ["model", "agent_name"],
    buckets=[0.1, 0.3, 0.5, 1.0, 2.0, 5.0]
)

tool_executions_total = Counter(
    "tool_executions_total",
    "Total tool executions",
    ["tool_name", "status"]
)

tokens_in_counter = Counter("llm_tokens_in", "Number of input tokens", ["agent_name"])
tokens_out_counter = Counter("llm_tokens_out", "Number of output tokens", ["agent_name"])

Every LLM call flows through model_invoke_with_metrics(), which times the inference and records token usage:

async def model_invoke_with_metrics(model, model_input, model_name, agent_name, config=None):
    with llm_inference_duration_seconds.labels(model=model_name, agent_name=agent_name).time():
        response = await model.ainvoke(model_input, config)
    record_token_usage(response, model_name, agent_name)
    return response

HTTP metrics are captured automatically by MetricsMiddleware, tracking request duration, counts, and status codes by endpoint.

Structured Logging

All logging uses structlog with strict conventions: lowercase_underscore event names and kwargs instead of f-strings, so logs are filterable and machine-parseable.

logger.info("chat_request_received", session_id=session.id, message_count=len(messages))

A LoggingContextMiddleware extracts session_id and user_id from the JWT on every request and binds them to the logging context:

class LoggingContextMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        clear_context()
        # Extract session_id from JWT
        payload = jwt.decode(token, settings.JWT_SECRET_KEY, algorithms=[settings.JWT_ALGORITHM])
        session_id = payload.get("sub")
        if session_id:
            bind_context(session_id=session_id)
        response = await call_next(request)
        clear_context()
        return response

The logging format switches automatically based on environment: colored console output in development, structured JSON in production.

9. The API Layer: FastAPI Endpoints and Streaming

The API provides both synchronous and streaming (SSE) endpoints for chat. The streaming endpoint uses FastAPI's StreamingResponse with an async generator:

@router.post("/chat/stream")
@limiter.limit(settings.RATE_LIMIT_ENDPOINTS["chat_stream"][0])
async def chat_stream(request, chat_request, session=Depends(get_current_session)):
    agent = await get_agent_example()

    async def event_generator():
        full_response = ""
        with llm_stream_duration_seconds.labels(model=settings.DEFAULT_LLM_MODEL, agent_name=agent.name).time():
            async for chunk in agent.agent_invoke_stream(
                chat_request.messages, session.id, user_id=session.user_id
            ):
                full_response += chunk
                response = StreamResponse(content=chunk, done=False)
                yield f"data: {json.dumps(response.model_dump())}\n\n"

        final_response = StreamResponse(content="", done=True)
        yield f"data: {json.dumps(final_response.model_dump())}\n\n"

    return StreamingResponse(event_generator(), media_type="text/event-stream")

The application wires everything together at startup using FastAPI's lifespan pattern:

@asynccontextmanager
async def lifespan(app: FastAPI):
    logger.info("application_startup", project_name=settings.PROJECT_NAME, version=settings.VERSION)
    await mcp_dependencies_init()
    yield
    await mcp_dependencies_cleanup()
    logger.info("application_shutdown")

app = FastAPI(title=settings.PROJECT_NAME, version=settings.VERSION, lifespan=lifespan)
setup_metrics(app)
setup_rate_limit(app)
app.add_middleware(LoggingContextMiddleware)
app.add_middleware(MetricsMiddleware)

10. MCP Integration: Model Context Protocol

The harness supports MCP (Model Context Protocol) for dynamically loading tools from external servers. MCP sessions are initialized at application startup and persist for the application lifetime.

class MCPSessionManager:
    async def initialize(self) -> Resource:
        self._exit_stack = AsyncExitStack()
        await self._exit_stack.__aenter__()

        tools, sessions = [], []
        for hostname in settings.MCP_HOSTNAMES:
            session = await self._exit_stack.enter_async_context(
                mcp_sse_client(hostname, correlation_id=generate_correlation_id())
            )
            session_tools = await load_mcp_tools(session)
            tools.extend(session_tools)
            sessions.append(session)

        self._resource = Resource(tools=tools, sessions=sessions)
        return self._resource

Key resilience features:

Multi-server support via MCP_HOSTNAMES_CSV environment variable
Automatic reconnection on ClosedResourceError with configurable retries
Graceful degradation -- if MCP servers are unavailable, the agent continues with built-in tools only
Correlation IDs on every MCP operation for traceability

MCP tools are loaded once and merged with built-in tools when the graph is compiled:

def _get_all_tools(self) -> list[BaseTool]:
    return self.tools + list(self.mcp_tools_by_name.values())

11. Evaluation Framework: LLM-as-Judge

The harness includes a metric-based evaluation framework that uses Langfuse traces as its data source and an LLM as the judge.

Metrics are defined as markdown files in src/evals/metrics/prompts/. Adding a new .md file auto-discovers it as a metric. Built-in metrics include: relevancy, helpfulness, conciseness, hallucination, and toxicity.

Each metric prompt instructs the evaluator LLM to score the output on a 0-to-1 scale:

Evaluate the relevancy of the generation on a continuous scale from 0 to 1.

## Scoring Criteria
A generation can be considered relevant (Score: 1) if it:
- Directly addresses the user's specific question or request
- Provides information that is pertinent to the query
- Stays on topic without introducing unrelated information

The evaluator fetches un-scored traces from the last 24 hours, runs each metric, and pushes scores back to Langfuse:

class Evaluator:
    async def run(self, generate_report_file=True):
        traces = self.__fetch_traces()
        for trace in tqdm(traces, desc="Evaluating traces"):
            for metric in metrics:
                input, output = get_input_output(trace)
                score = await self._run_metric_evaluation(metric, input, output)
                if score:
                    self._push_to_langfuse(trace, score, metric)

The framework uses OpenAI's structured outputs to ensure the LLM returns a valid score and reasoning:

class ScoreSchema(BaseModel):
    score: float  # 0-1
    reasoning: str

Reports are saved as JSON with per-metric averages and per-trace breakdowns. Run evaluations with:

make eval              # Interactive mode
make eval-quick        # Default settings, no prompts
make eval-no-report    # Skip report generation

12. Configuration and Environment Management

The harness uses environment-specific configuration files loaded in priority order:

.env.{environment}.local  (highest priority, git-ignored)
.env.{environment}
.env.local
.env                      (lowest priority)

Four environments are supported -- development, staging, production, and test -- each with automatic overrides:

env_settings = {
    Environment.DEVELOPMENT: {
        "DEBUG": True, "LOG_LEVEL": "DEBUG", "LOG_FORMAT": "console",
        "RATE_LIMIT_DEFAULT": ["1000 per day", "200 per hour"],
    },
    Environment.PRODUCTION: {
        "DEBUG": False, "LOG_LEVEL": "WARNING",
        "RATE_LIMIT_DEFAULT": ["200 per day", "50 per hour"],
    },
}

Explicit environment variables always take precedence over these defaults, so you can override any setting per-deployment without code changes.

13. DevOps: Docker, Compose, and CI/CD

The Dockerfile uses a slim Python base with a non-root user for security:

FROM python:3.13.2-slim

COPY pyproject.toml .
RUN uv venv && . .venv/bin/activate && uv pip install -e .

COPY . .
RUN useradd -m appuser && chown -R appuser:appuser /app
USER appuser

ENTRYPOINT ["/app/scripts/docker-entrypoint.sh"]
CMD ["/app/.venv/bin/uvicorn", "src.app.main:app", "--host", "0.0.0.0", "--port", "8000"]

Docker Compose provides the full monitoring stack:

PostgreSQL with pgvector -- Database for users, sessions, checkpoints, and vector memory
Prometheus -- Scrapes /metrics from the FastAPI app
Grafana -- Pre-configured dashboards for API latency, LLM inference duration, token usage, and rate limiting
cAdvisor -- Container-level resource metrics

The Makefile serves as the single CLI entrypoint:

make dev                              # Start development server with uvloop
make test                             # Run pytest
make eval                             # Run evaluations
make docker-build-env ENV=production  # Build Docker image
make docker-compose-up ENV=development # Full stack
make lint                             # Ruff check and format

14. Building Your Own Agent

Every agent lives in its own directory. Start by choosing the architecture that fits your use case -- refer to the comparison in section 2 -- then follow the pattern from the matching reference agent.

Step 1: Create the agent directory

src/app/agents/my_agent/
  __init__.py          # Factory function
  agent.py             # Agent class with graph definition
  system.md            # System prompt template (or AGENTS.md for ReAct)
  tools/
    __init__.py        # Export tools list
    my_tool.py         # Custom tool implementations

Step 2: Choose your architecture

Custom graph workflow (like chatbot) -- Define a StateGraph with explicit nodes and edges. Best when you want full control over execution flow.

Multi-agent subgraphs (like deep research) -- Build separate compiled subgraphs and compose them as nodes in a parent graph. Best for complex tasks with decomposable sub-problems.

ReAct with Deep Agents (like text-to-SQL) -- Use create_deep_agent with markdown memory files, skill directories, and tool kits. Best for domain-specific tool use where the agent should decide its own execution order.

Step 3: Define the system prompt

For graph-based agents, system.md supports {long_term_memory} and {current_date_and_time} placeholders, plus any custom kwargs:

# Name: {agent_name}
# Role: A domain-specific assistant

Your specific instructions here.

## What you know about the user
{long_term_memory}

## Current date and time
{current_date_and_time}

For ReAct agents, use AGENTS.md to define the agent's identity, rules, and planning strategies in natural language.

Step 4: Build the agent class

Follow the pattern from the reference agent that matches your chosen architecture. At minimum, implement agent_invoke (and optionally agent_invoke_stream for SSE support). Apply guardrails either as graph nodes (custom graph / subgraph approaches) or as wrapper logic (ReAct approach):

class MyAgent:
    def __init__(self, name, tools, checkpointer):
        self.name = name
        self.tools = tools
        self.checkpointer = checkpointer

    async def compile(self):
        graph_builder = await self._create_graph()
        self._graph = graph_builder.compile(checkpointer=self.checkpointer, name=self.name)

Step 5: Wire it to an API endpoint

Create a route in src/app/api/v1/ following the chatbot pattern:

@router.post("/chat", response_model=ChatResponse)
@limiter.limit(settings.RATE_LIMIT_ENDPOINTS["my_endpoint"][0])
async def chat(request: Request, chat_request: ChatRequest, session=Depends(get_current_session)):
    agent = await get_my_agent()
    result = await agent.agent_invoke(chat_request.messages, session.id, user_id=session.user_id)
    return ChatResponse(messages=result)

The harness handles everything else: auth, memory retrieval and update, checkpointing, metrics, tracing, and guardrails.

15. Production Readiness Checklist

Here is a summary of the production concerns addressed by the harness:

Concern	Implementation
Authentication	JWT tokens with session-per-conversation model
Input guardrails	Content filter, prompt injection detection, PII blocking
Output guardrails	PII redaction, LLM-based safety evaluation
Long-term memory	mem0 + pgvector, per-user, non-blocking updates
Context management	Tool-result eviction to disk, two-stage conversation summarization
State persistence	LangGraph AsyncPostgresSaver with automatic checkpointing
Observability	Langfuse tracing, Prometheus metrics, structlog logging
Rate limiting	Per-endpoint limits via slowapi, configurable per environment
Error handling	Retries with exponential backoff, graceful degradation
Streaming	SSE endpoints with token-by-token streaming
Evaluation	LLM-as-judge with auto-discovered metrics and Langfuse integration
Deployment	Docker with non-root user, Compose with full monitoring stack
MCP	Multi-server tool loading with reconnection and graceful fallback
Agent architectures	Custom graph, multi-agent subgraphs, and ReAct -- pick the pattern that fits

The gap between a working agent and a production agent is not the LLM logic -- it is everything around it. A harness approach lets you solve these problems once, then focus on what matters: the agent's behavior. Whether you need a simple chatbot with a tool loop, a multi-agent research system with parallel delegation, or an autonomous ReAct agent reasoning over a database, the same infrastructure wraps them all.

The full source code is available on GitHub. Clone it, swap in your agent, and ship.

A Deep Dive into Deep Agent Architecture for AI Coding Assistants

Alexsandro Souza — Thu, 22 Jan 2026 18:07:15 +0000

Traditional AI agents operate with a fundamental limitation: they're single agents trying to handle everything at once. While they can be impressively capable for straightforward tasks, they struggle when faced with complex, multi-step challenges that require sustained reasoning, strategic planning, and coordinated execution.

The problems are threefold:

Limited Context Memory: Even with extended context windows, single-agent systems quickly exhaust their memory when dealing with multi-step tasks. Every file read, every search result, every intermediate finding consumes precious context space, leaving little room for actual reasoning.
Lack of Specialization: A single agent must simultaneously handle strategic planning, investigation, implementation, and verification. This jack-of-all-trades approach often leads to shallow execution. The same agent analyzing system architecture is also writing code and running tests, with no clear separation of concerns or focused expertise.
No Compound Intelligence: Traditional agents can't meaningfully build on their own discoveries. Each action happens in isolation, with no accumulation of knowledge. There's no concept of "this investigation discovered X, now another agent can use that knowledge to do Y more effectively."

Enter Deep Agent architecture - a fundamentally different approach that addresses these limitations through multi-agent collaboration, forced specialization, and a shared Context Store. This is the same advanced pattern powering production tools like Claude Code, Manus, and Deep Research.

This article explores an experimental implementation built entirely from scratch (no frameworks!) to reveal what's actually under the hood of these systems. By building without abstractions, we can see exactly how each component works - from context management to agent coordination to action parsing.

The key innovation? A shared Context Store that enables "compound intelligence" where the system gets smarter as it works:

Knowledge Accumulation: Every discovery becomes a permanent building block
No Redundant Work: Agents never rediscover the same information
Focused Execution: Each agent receives only the contexts it needs

You'll learn how multi-agent collaboration creates compound intelligence, how architectural constraints can improve system design, and how the Context Store transforms the way AI agents work together - all by examining a real implementation built from the ground up.

System Overview: Learning Through Agentic Code Assistant

To understand Deep Agent architecture in practice, we'll dive deep into Agentic Code Assistant - an experimental, open-source implementation built entirely from scratch without using any agent frameworks. This from-the-ground-up approach reveals exactly how Deep Agent systems work internally, making visible the patterns that frameworks typically hide.

By examining this real system, we'll explore how Deep Agent architecture patterns work in practice: multi-agent coordination, context management, forced delegation, and the intricate details that make these systems effective.

Throughout this article, we'll use Agentic Code Assistant as our reference implementation to learn the principles, patterns, and practical considerations that power production Deep Agent systems.

What Makes Deep Agents Different

Unlike traditional agents that operate in short loops or perform simple tool calls, Deep Agents:

Plan their actions strategically before execution
Manage evolving context through persistent knowledge stores
Delegate subtasks to specialised sub-agents with focused expertise
Maintain state across long, complex interactions
Build compound intelligence through knowledge sharing

Core Capabilities

The system transforms high-level coding requests into executed implementations through four key capabilities:

Strategic Task Decomposition When you submit a complex task like "Add authentication to the API," the system doesn't immediately start coding. Instead, it analyzes the request, breaks it into logical subtasks (investigation, planning, implementation, verification), and executes them systematically.
Sustained Reasoning Across Long Interactions Multi-step tasks that would overwhelm a single agent's context window become manageable through structured delegation and knowledge passing. The system maintains coherent progress even across dozens of turns.
Knowledge Sharing Through Context Store Every discovery - from "the user model is in /models/user.py" to "authentication uses JWT tokens" - becomes a reusable knowledge artifact. Subsequent agents in the workflow access this accumulated knowledge without redundant investigation.
Built-in Verification Implementation isn't the end. The system systematically verifies changes through test execution and validation, ensuring quality before considering a task complete.

The Typical Workflow

Here's how the system handles a real request:

This workflow demonstrates the system's core innovation: knowledge flows through the agents via distilled contexts, with each step building meaningfully on previous discoveries. The Coder doesn't waste time searching for the user model

the Explorer already found it and reported it as a context. The verification Explorer receives implementation contexts from the Coder, enabling focused testing. Importantly, the Orchestrator never sees the Explorer's full investigation history or the Coder's entire working context - only the relevant, distilled knowledge artifacts. This keeps the orchestration layer lean and focused on strategy.

Architecture Deep-Dive: How Deep Agent Systems Work

Now let's examine the technical architecture that enables this compound intelligence. The system is built on three specialised agents, a persistent knowledge store, and carefully designed architectural constraints.

Multi-Agent Design: Specialisation Through Separation

The system employs three distinct agent types, each with specific responsibilities and access levels:

The Orchestrator (Lead Architect)

The Orchestrator is the strategic brain of the system - but with a crucial limitation: it has no direct access to code . This forced delegation pattern (which we'll explore in detail shortly) shapes its entire behavior.

Responsibilities:

Analyze incoming tasks and create strategic execution plans
Decompose complex requests into focused subtasks
Manage the Context Store (decide what knowledge to create and when to use it)
Track all tasks via the Task Store
Coordinate subagent execution
Synthesize results and verify completion

Key Constraint: Cannot read, write, or modify code directly. Must delegate all code-related work to subagents.

Why This Matters: By removing direct code access, the architecture forces proper task decomposition. The Orchestrator can't take shortcuts - it must think strategically about what information is needed, who should gather it, and how to coordinate execution. This constraint encourages better system design.

The Explorer (Read-Only Investigator)

Explorers are investigation and verification specialists with read-only access to the codebase.

Responsibilities:

Investigate codebase structure and patterns
Search for specific implementations or conventions
Verify implementations through test execution
Gather information and report contexts containing discoveries
Validate that changes work as expected

Key Constraint: Read-only access - can execute code and tests, but cannot modify files.

Why This Matters: Separation of investigation from implementation prevents the common anti-pattern of "quick fixes while exploring." Explorers focus purely on understanding and verification. They might read dozens of files during investigation, but report back only the essential contexts - the distilled knowledge. This produces thorough, accurate knowledge artifacts without polluting the Orchestrator's context with raw investigation data.

The Coder (Implementation Specialist)

Coders are implementation specialists with full read-write access to the codebase.

Responsibilities:

Implement features and bug fixes
Make code modifications (create, edit, delete files)
Run tests to validate their own implementations
Report changes with contexts describing what was implemented and how
Self-validate before reporting back

Key Constraint: Operates with specific implementation instructions from the Orchestrator, with all necessary context pre-loaded.

Why This Matters: Coders receive everything they need upfront - architectural knowledge from Explorers, clear instructions from the Orchestrator, and relevant contexts from the Context Store. They can focus entirely on implementation quality. When done, they report back with contexts containing implementation details, not their entire working history. This keeps the knowledge base focused on outcomes, not process.

The Context Store Innovation: The Key to Compound Intelligence

The Context Store is the system's most important innovation. It's a shared knowledge base that enables true compound intelligence through knowledge accumulation and reuse during task execution.

The Problem It Solves

Traditional AI agents suffer from "knowledge amnesia":

Repetitive Discovery: Each agent rediscovers the same information, wasting context and time
Wasted Resources: Precious tokens spent rediscovering known information
Context Explosion: Every agent carries full context history, quickly exhausting available context windows
No Learning: Knowledge gained by one agent is lost to the next

The Context Store solves this by making every discovery shareable and reusable. Sub-agents maintain their full context only while running, then report back only the relevant contexts they discovered. This keeps the Orchestrator's context lean while enabling compound intelligence.

How It Works

The system treats knowledge as persistent artifacts with a clear lifecycle:

Context Creation The Orchestrator identifies what knowledge is needed and creates named context "slots":

context_refs:
    - user_model_schema
    - database_configuration
    - authentication_patterns

Knowledge Discovery Subagents (typically Explorers) investigate and populate these contexts. While running, sub-agents maintain their full context including all file reads, searches, and intermediate findings. However, when they complete, they report back only the distilled, relevant contexts:

{
    "id": "authentication_patterns",
    "content": "The codebase uses JWT tokens for authentication. Implementation in /middleware/auth.py. User model includes email and hashed_password fields. Token generation uses jose library with HS256 algorithm.",
    "metadata": {
        "created_by": "explorer",
        "task_id": "investigate_auth"
    }
}

Centralized Storage Contexts are stored centrally, indexed by ID, and available to all agents during the task execution. This means:

Sub-agents work with full context while running (investigating, reading files, etc.)
Only relevant, distilled contexts are reported back to the Orchestrator
The Orchestrator's context stays lean and focused on strategy
Future sub-agents receive only the specific contexts they need

Context Injection When launching new subagents, the Orchestrator injects relevant contexts:

task_create:
    agent_name: coder
    title: Implement auth middleware
    context_refs:
        - authentication_patterns  # Reused from previous task
        - user_model_schema        # Reused from previous task
        - api_route_structure      # Newly discovered

Knowledge Building Each task adds to the knowledge base. The first task discovers auth patterns. The second task implements auth and adds implementation details. The third task adds test patterns. Each builds on the previous.

Context Types

The system supports multiple context types, each capturing different knowledge:

Code Context: File contents, function signatures, class hierarchies
Architecture Context: System design, dependencies, patterns in use
Test Context: Test results, coverage information, verification outcomes
Bug Context: Error analysis, root causes, reproduction steps
Implementation Context: Changes made, approach used, trade-offs considered

Real-World Impact

Consider this scenario within a complex task execution:

Main Task: "Add comprehensive error tracking with logging to the API"

Key Observations:

Explorer's full context (all 20 files read, searches performed) is discarded - only the distilled contexts remain
New Explorer receives logging_config context (no need to rediscover)
Coder receives logging_config + error_patterns + log_patterns contexts - zero context wasted on rediscovering infrastructure
Final Explorer receives all relevant contexts for focused verification

This is compound intelligence in action: knowledge accumulates during execution, each agent builds on previous discoveries, and the Orchestrator's context remains focused on orchestration rather than drowning in raw investigation data.

The Forced Delegation Pattern: Architectural Constraint as Design Principle

One of the most interesting aspects of this architecture is what the Orchestrator cannot do: access code directly. This isn't a limitation - it's an intentional design constraint that dramatically improves system behavior. The Orchestrator agent has no access to file reading, writing, or editing tools. If it needs to know what's in a file, it must delegate to an Explorer. If it wants code modified, it must delegate to a Coder. There are no shortcuts.

Why This Design Choice

Forces Proper Task Decomposition Without shortcuts available, the Orchestrator must think systematically:

What information do we need? (Launch Explorer)
What should be built? (Plan based on Explorer findings)
How should it be implemented? (Provide clear instructions to Coder)
Is it correct? (Launch Explorer to verify)

Quick fixes and ad-hoc changes become impossible, encouraging methodical execution.

Encourages Strategic Thinking When you can't implement directly, you must think strategically about delegation:

What contexts does the Coder need?
Has an Explorer already discovered this information?
Should verification happen before or after the next step?

This leads to better-planned, more coherent implementations.

Creates Clear Separation of Concerns Strategy and implementation become truly separate:

Orchestrator: "What needs to happen and why"
Coder: "How to implement it"
Explorer: "What exists and whether it works"

Each agent can focus on its expertise without role confusion.

Prevents Anti-Patterns Common failure modes become impossible:

Can't mix investigation with hasty implementation
Can't skip verification because "it should work"
Can't bypass proper knowledge sharing to save time

Impact on System Behavior

In practice, this constraint produces notably more systematic implementations. The forced delegation version is more verbose, but produces more thorough, better-verified implementations.

Stateless Turn-Based Execution

The system follows a clear, reproducible execution pattern:

Each Turn:

LLM receives current state + history + feedback from previous turn
LLM outputs actions (XML-wrapped YAML)
System executes all actions in the turn
System provides feedback (success/error/results)
Cycle repeats until task completion or turn limit

This stateless design means agents don't carry hidden state - everything is explicit in the turn history.

Technical Implementation: How the System Actually Works

Now that we understand the architecture conceptually, let's examine how it's actually implemented.

The Action System: How Agents Communicate

Agents communicate intentions through structured actions using a hybrid XML/YAML format that's both LLM-friendly and human-readable.

Action Format

Actions are XML tags containing YAML-structured parameters:

<task_create>
    agent_name: explorer
    title: Find authentication patterns
    description: |
    Investigate the codebase to discover:
    - Location of user model
    - Authentication library in use
    - Existing auth middleware
    - Database configuration for users
    context_refs:
    - database_config
    - user_model_schema
    auto_launch: true
</task_create>

Why This Format?

XML tags are easy for LLMs to generate reliably
YAML content is human-readable and supports complex structures
Pydantic validation ensures type safety and catches errors early

Actions by Agent Type

Each agent type has access to specific actions based on its role:

Orchestrator Actions:

task_create: Create new subagent tasks
launch_subagent: Launch created tasks
add_context: Add knowledge to Context Store
finish: Mark task complete and return to user

Explorer Actions:

file (read mode): Read file contents
file (metadata mode): Get file information
search (grep/glob): Search for code patterns
bash: Execute commands (read-only)
todo: Track investigation progress
scratchpad: Take notes
report: Report findings with contexts

Coder Actions:

file (read/write/edit/multi_edit): Full file manipulation
bash: Execute commands (including installations)
search (grep/glob): Find code to modify
todo: Track implementation progress
scratchpad: Document approach
report: Report implementation with contexts

Note the access level differences: Explorers can only read files, while Coders can modify them. The Orchestrator can't access files at all.

Action Parsing Pipeline

The system processes actions through a multi-stage pipeline:

Pipeline Stages:

XML Extraction: Parse LLM response to find XML tags
YAML Parsing: Extract and parse YAML content from each tag
Pydantic Validation: Validate structure and types
Handler Routing: Route to appropriate specialized handler
Execution: Execute the action
Feedback Generation: Create structured feedback for next turn

Example Flow:

# LLM outputs:
"""
<file_read>
path: /src/auth.py
</file_read>
"""

Communication Protocol: Reports and Context Flow

Sub-agents communicate results through structured reports that contain contexts - the distilled knowledge from their execution:

{
    "contexts": [
        {
            "id": "auth_middleware_location",
            "content": "Authentication middleware is located at /src/middleware/auth.py. It uses JWT tokens with HS256 algorithm..."
        },
        {
            "id": "test_results",
            "content": "Ran authentication test suite: 15 tests, all passing. Verified: token generation, validation, expiration..."
        }
    ],
    "comments": "Investigation complete. All authentication code documented and verified.",
    "meta": {
        "num_turns": 5,
        "total_input_tokens": 12450,
        "total_output_tokens": 3820,
        "model": "openai/gpt-5"
    }
}

This reporting mechanism is crucial for context efficiency:

While the sub-agent ran, it maintained full context including:

All files it read
All search results
All intermediate findings
All bash command outputs

But when reporting back, it distills this into only the relevant contexts. The Orchestrator never sees the raw investigation data - only the refined knowledge.

Knowledge Flow:

Key Points:

Sub-agent works with full context during execution
Sub-agent reports only distilled contexts when complete
Context Store persists contexts centrally
Orchestrator remains lean, working only with strategic information
Future agents receive only the specific contexts they need

This creates compound intelligence without context explosion. Each agent builds on previous discoveries, but the system doesn't accumulate unnecessary detail.

Project Structure: Code Organization

The implementation is organized into clear modules:

src/
├── core/                        # Core system implementation
│   ├── agent/                  # Base agent classes and task definitions
│   │   ├── agent.py           # Abstract Agent base class
│   │   └── agent_task.py      # AgentTask data model
│   ├── orchestrator/           # Orchestrator implementation
│   │   ├── orchestrator_agent.py
│   │   ├── orchestrator_hub.py
│   │   └── factory.py         # Agent factory
│   ├── context/                # Context Store implementation
│   │   └── context_store.py
│   ├── task/                   # Task management
│   │   ├── task_store.py
│   │   └── task_manager.py
│   ├── action/                 # Action system
│   │   ├── action_handler.py
│   │   ├── action_parser.py
│   │   └── handlers/          # Specialized action handlers
│   ├── file/                   # File operations
│   │   └── file_manager.py
│   ├── search/                 # Search operations
│   │   └── search_manager.py
│   ├── bash/                   # Command execution
│   │   └── docker_executor.py
│   └── llm/                    # LLM client
│       └── litellm_client.py
├── system_msgs/                # Agent system prompts v0.1
│   └── md_files/
│       ├── orchestrator_sys_msg_v0.1.md
│       ├── explorer_sys_msg_v0.1.md
│       └── coder_sys_msg_v0.1.md
└── misc/                        # Logging and utilities
    └── pretty_log.py

test/                            # Test scripts and examples
├── real_simple_task.py         # Simple file creation example
└── real_advanced_task.py       # Complex API server example

Design Principles:

Clear separation: Each module has a single, focused responsibility
Dependency injection: Components receive dependencies explicitly
Type safety: Pydantic models throughout for validation
Testability: Components can be tested in isolation

Docker-Based Execution Environment

The system executes code in isolated Docker containers to ensure:

Safety: No risk to the host system
Reproducibility: Consistent environment across runs
Isolation: Each test run is independent

# From real_simple_task.py
executor = DockerExecutor(container_name)
executor.execute("mkdir /workspace")
executor.execute("cd /workspace")

Agents execute code in this container

All file operations happen inside Docker

This makes the system safe to experiment with - agents can install dependencies, run code, and make changes without affecting your actual development environment.

Practical Applications: Real-World Examples

Theory is interesting, but let's see the system in action with real examples from the codebase.

Example 1: Simple File Creation

The simplest example demonstrates the core workflow:

Task: "Create a file called 'hello.txt' with content 'Hello world! Ending with a newline.'"

What Happens:

# From test/real_simple_task.py
instruction = (
    "Create a file called 'hello.txt' with content 'Hello world! Ending with a newline.'"
)
agent_task = AgentTask(task_id="test_task", instruction=instruction)
result = orchestrator.run_task(agent_task, max_turns=15)

Execution Flow:

Orchestrator receives task
Orchestrator creates task:
Coder executes:
Coder reports back:
Orchestrator verifies:
Result:

Even for this simple task, the system follows proper delegation patterns. The Orchestrator doesn't create the file itself - it delegates to a specialist.

Example 2: Complex API Server Implementation

A more realistic example shows the system handling multi-step complexity:

Task: "Create and run a server on port 3000 with GET /fib endpoint. It should accept a query param /fib?n={number} and return the nth Fibonacci number as a JSON object. Handle error cases for missing/invalid parameters."

# From test/real_advanced_task.py
instruction = """Create and run a server on port 3000 that has a single GET endpoint: /fib.

It should expect a query param /fib?n={some number} and return the nth Fibonacci number
as a JSON object with a key 'result'.

If the query param is not provided, it should return a 400 Bad Request error.
If the query param is not an integer, it should return a 400 Bad Request error.

Automatically choose to install any dependencies if it is required.

Execution Flow:

Orchestrator analyzes task:
Orchestrator launches Explorer:
Explorer investigates:
Explorer reports:
Orchestrator synthesizes plan:
Orchestrator launches Coder with contexts:
Coder implements:
Coder reports implementation:
Orchestrator verifies:
Result:

Key Observations:

Strategic Investigation: Explorer investigated environment before implementation
Context Reuse: Coder received environment information without rediscovering it
Systematic Implementation: Breaking task into subtasks (install, implement, test)
Built-in Verification: Explorer tested multiple cases to ensure correctness
Knowledge Accumulation: Contexts created (fibonacci_api_implementation, api_error_handling) available for future tasks

Use Case Patterns

The system excels in several common software development scenarios:

1. Feature Implementation

Pattern: Investigate → Plan → Implement → Verify

Example: "Add user profile endpoints to API"

Explorer investigates: existing user model, database schema, API patterns
Orchestrator plans: endpoints needed, validation requirements
Coder implements: profile GET/PUT endpoints with validation
Explorer verifies: tests all endpoints, edge cases

Benefit: Implementations follow existing patterns and conventions

2. Bug Investigation and Fix

Pattern: Reproduce → Investigate → Diagnose → Fix → Verify

Example: "Fix authentication failing for expired tokens"

Explorer reproduces: creates test case showing failure
Explorer investigates: traces auth flow, identifies token validation issue
Orchestrator diagnoses: token expiration check missing
Coder fixes: adds expiration validation
Explorer verifies: all auth tests passing, expired tokens rejected

Benefit: Systematic approach ensures root cause is addressed, not just symptoms

3. Code Refactoring

Pattern: Understand → Plan → Refactor → Validate

Example: "Refactor database connection handling"

Explorer maps: all current database connection code
Orchestrator plans: centralized connection management approach
Coder refactors: creates connection pool, updates all usage sites
Explorer validates: all tests passing, no broken connections

Benefit: Strategic planning prevents breaking changes

4. Test Generation

Pattern: Understand → Identify Gaps → Generate → Validate

Example: "Add comprehensive tests for authentication"

Explorer understands: reads all auth code, identifies untested paths
Orchestrator identifies: missing edge case tests (expired tokens, invalid formats, etc.)
Coder generates: comprehensive test suite
Explorer validates: runs tests, confirms coverage improvement

Benefit: Tests are comprehensive and understand the actual implementation

Getting Started: Try It Yourself

Want to run the system and see it in action? Check the Github's repository to see how to get started.

Conclusion: Building Toward Truly Capable AI Development Tools

Let's step back and examine what this architecture teaches us about the future of AI-powered software development.

Key Takeaways

Specialization Creates Compound Intelligence

The system demonstrates that multiple specialized agents can achieve more than a single generalist agent. By dividing responsibilities - strategy (Orchestrator), investigation (Explorer), and implementation (Coder) - each agent can focus on its expertise. The whole becomes greater than the sum of its parts.

The Context Store Enables Compound Intelligence Without Context Explosion

The Context Store is transformative. Traditional agents rediscover the same information repeatedly, wasting tokens and time. The innovation here is dual: knowledge accumulates (each agent builds on previous discoveries), but context stays lean (sub-agents report only distilled contexts, not their full working context). This enables compound intelligence without overwhelming the system with detail.

Architectural Constraints Can Improve Design

The forced delegation pattern - preventing the Orchestrator from directly accessing code - seems like a limitation but produces better outcomes. By removing shortcuts, the architecture encourages systematic thinking, proper decomposition, and thorough verification. Sometimes constraints clarify thinking.

Stateless Turn-Based Execution Provides Clarity

The explicit turn structure (LLM → Actions → Execution → Feedback) creates reproducible, debuggable behavior. Each turn has clear inputs and outputs. This transparency makes the system easier to understand, debug, and improve.

Why This Matters

This project demonstrates production-grade agent architecture patterns in an accessible, educational format. Built entirely from scratch without frameworks, it reveals the actual mechanics behind systems like Claude Code, Manus, and Deep Research - showing not just what they do, but how they do it.

More importantly, it bridges the gap between research and practical tools. The concepts here - multi-agent collaboration, context-efficient knowledge sharing, forced delegation patterns - aren't just interesting theoretically. They solve real problems that single-agent systems face: context explosion, lack of specialization, and inability to build compound intelligence. By building from scratch, we can see exactly how each pattern addresses these challenges.

Future Potential

The architecture is designed for extensibility. Imagine additional specialized agents:

Security Agent: Focused on vulnerability detection and secure coding patterns
Documentation Agent: Creates and maintains comprehensive documentation
Performance Agent: Profiles code and implements optimizations
Test Agent: Specialized in test generation and coverage analysis

Each would contribute to the Context Store during task execution, building an increasingly sophisticated understanding through their specialized investigations.

The Task Store could enable even longer-running tasks with failure recovery. If a task fails, the system could analyze the failure, adjust the approach, and retry - learning from mistakes within the same execution.

The Context Store could become a true knowledge graph, with contexts linking to related contexts, enabling semantic queries during task execution: "Find all contexts related to authentication" or "Show me everything discovered about the user model."

Call to Action

The project is open-source under Apache 2.0 license. Built from scratch without frameworks, the codebase is designed to be readable and extensible - perfect for learning how these systems actually work. Whether you're:

A developer curious about agent architectures and what's under the hood
A researcher exploring multi-agent systems and coordination patterns
An engineer building AI-powered tools who wants to understand the internals
A student learning about AI application architecture from first principles

There's value in exploring the code, trying the examples, and understanding how production Deep Agent systems work at a fundamental level.

GitHub Repository: https://github.com/yourusername/agentic-code-assistant

Try the examples. Read the code. Experiment with different LLM models. Contribute improvements. Build additional agents. The goal is education and exploration.

The future of AI systems isn't single agents trying to do everything. It's specialised agents working together, sharing distilled knowledge through efficient context management, and building compound intelligence without context explosion. This project demonstrates that future - today.

Happy coding.

How to build your own AI Coding assistant

Alexsandro Souza — Thu, 13 Nov 2025 10:01:19 +0000

Excited to share an experimental project I've been working on: an agentic AI code assistant built entirely from scratch (no frameworks!).

Do you wanted to move beyond simple agent loops and explore the "Deep Agent" architecture style—the same advanced pattern powering tools like Claude Code, Manus and Deep Research.

This architecture excels at complex, multi-step tasks. My implementation features:

A central Orchestrator Agent to analyse, plan, and delegate.
A specialised Explorer Agent to investigate the codebase.
A specialised Coder Agent to implement changes.

The key innovation? A shared Context Store.
This central "memory" allows agents to share findings, eliminating redundant work and creating a "compound intelligence" where the system gets smarter as it works. It's a fascinating model for strategic delegation and sustained reasoning.

Check out the Github project to learn more

Saving and Investing Is Not Enough

Alexsandro Souza — Thu, 06 Nov 2025 10:02:42 +0000

Traditional financial advice is usually limited to saving and investing your money. While important, that's only one piece of the puzzle.

Building true wealth involves far more than simply saving, starting a business, or acquiring the usual income-generating assets like bonds, stocks, or real estate. In this post, we're skipping the basics and diving straight into the concepts that create real financial momentum.

We'll explore one foundational mindset and two powerful strategies:

The Mindset: Understanding money as a tool.
Strategy 1: Using other people's money (Leverage).
Strategy 2: The science of tax efficiency.

It's time to learn the simple truths that make the hard work worth it.

The Foundational Mindset: Money is a Tool

To truly leverage the system, you must first understand how money works. Money is a remarkable tool for universal interaction built entirely on confidence. This shared trust allows it to act as a universal language for value, but like any tool, it has both a supply and a price.

The supply is controlled by central banks. When they increase it, the value of the money in your account decreases—a process known as inflation. The bills in your wallet only have power because we all collectively agree they do. When that trust erodes, a currency can rapidly lose its value (hyperinflation).

Lesson 1: Build your wealth in strong, stable currencies. If you live in a country with a volatile currency (like Argentina or Turkey, which have seen >80% value loss), prioritize owning hard assets over holding cash.

Money also has a price. You can walk into a bank and "buy" money; the cost is the interest you pay to borrow it.

Lesson 2: It is absolutely crucial to pay as little as possible for the money you borrow.

Understanding this price is the key to our first major strategy.

Strategy 1: Master Financial Leverage

If money has a price, your goal is to pay as little for it as possible and use it to acquire assets that generate a higher return. This is the art of leverage.

Even if you have your own capital, you can (and should) still leverage other people's money, so long as its cost is low and you're able to generate enough income to pay it back.

There are many ways to raise "interest-free" or "low-interest" money:

Paying bills in installments
Delaying tax payments until the due date
Using 0% interest credit card offers
Finding an investment partner
Crowdfunding
Getting paid in advance and delaying payments

Airlines, for example, master this last point. They collect your ticket money months before your trip but may not pay for the fuel for that trip until a month after, effectively giving them a massive, interest-free loan to run their business.

When an interest-free loan isn't an option, the next stop is a bank. The key is to secure the lowest possible interest rate. You achieve this by increasing your credit score and proving your capacity to repay—often by using your existing assets as collateral.

This leads to a crucial lesson: Your goal is to accumulate assets not just to generate income, but also to use as leverage for borrowing.

Let's break this down.

Unsecured Loan: You ask for a personal loan. The bank only has your promise to pay it back. This is high-risk for them, so they charge a high interest rate (e.g., 15%).
Asset-Backed Loan: You ask for a loan against your stock portfolio. The bank can claim your assets if you default. This is low-risk for them, so they give you a much lower rate (e.g., 4%).

This is how the wealthy access cash: they don't sell their appreciating assets; they borrow against them.

Imagine your investments are returning 8% annually, but you can borrow against them for 4%. You're effectively making money on the loan. Better yet, since you never sold the asset, you don't trigger a capital gains tax (more on that in a moment) and your original wealth continues to compound.

Now that you can access cheap capital, it's time to choose an investment that will give you a return above your loan's interest rate. Real estate is a classic example. You might use a £50,000 deposit to buy a £500,000 house. The rental income covers the mortgage and expenses, and the rest is cash flow. Meanwhile, you're building equity and increasing your creditworthiness for the next deal—all by using the bank's money for 90% of the purchase.

Strategy 2: The Science of Tax Efficiency

Governments can be one of the biggest hurdles to building wealth. In many countries, taxes can claim over 50% of your income. Understanding how to legally minimize your tax burden is not just an advantage; it's a necessity.

Here are several powerful strategies to improve your tax efficiency.

Leverage Geographic Mobility

For many, the ability to work from anywhere is a reality. One of the most direct ways to lower your tax bill is to establish tax residency in a country with a more favorable system. Many countries actively compete for talent by offering special tax incentives, like Spain's "Beckham Law." Even if you can't move now, you can plan. For instance, you could delay selling assets until you become a tax resident in a location with low or zero capital gains taxes.

Borrow, Don't Sell

This is a powerful tax strategy we touched on earlier. Selling an appreciating asset is often a last resort. Why? Because selling triggers a taxable event, immediately handing over a chunk of your gains to the government.

Instead, by using your portfolio as collateral for a low-interest loan, you get the cash you need without the tax bill. This is a triple win:

Your original investment continues to grow and compound.
The loan interest is often lower than your investment returns.
The capital gains tax you deferred effectively stays invested, compounding for you.

Maximize Tax-Sheltered Accounts

One of the most common and powerful tax shelters is a private pension. In most countries, contributions are made pre-tax. If you're a 40% taxpayer, every £100 you contribute to your pension only costs you £60 from your take-home pay. The government essentially pays the other £40. That's an immediate 40% saving, not to mention the decades of tax-free growth.

Use a Corporate Structure

As an employee, your tax-saving options are limited. But if you can receive your income through a limited company, you unlock a new level of financial strategy. This isn't just about a lower corporation tax rate; it's about control.

Your company can deduct business expenses, reducing its taxable profit. More importantly, you can decide how and when you get paid. You can draw a small, tax-efficient salary and take the rest as dividends at a time that suits your financial picture, helping you stay in lower tax brackets.

Actively Seek Credits and Relief

Finally, don't forget the ground game. Actively seek out every tax credit, relief, and social welfare option available in your country. You might be surprised by the valuable benefits you're leaving on the table. A few hours of research could be the most profitable time you spend all year.

Conclusion: Simple Doesn't Mean Easy

Building wealth isn't a secret; it's a system. We've seen that this system goes far beyond just "earning more" or "saving more."

It requires a new way of thinking:

Treat money as a tool: Understand its price (interest) and its stability (inflation).
Master leverage: Use other people's capital to acquire assets, ensuring your returns are always higher than your cost.
Play defense: Master tax efficiency to legally keep more of what you earn, letting it compound for your benefit, not the government's.

These strategies are simple to understand, but they are not easy to execute. They involve taking calculated risks, continuous learning, and hard work. But by understanding the system, you're no longer just working hard—you're working smart.

Unveiling the Magic Behind Autonomous AI Agents

Alexsandro Souza — Tue, 08 Oct 2024 09:57:12 +0000

Have you ever wondered what powers AI agents and allows them to think, reason, and act on their own? It all comes down to a powerful technique called ReAct (Reasoning and Acting), which forms the foundation of popular frameworks like AutoGPT, CrewAI, and LangChain.

In this post, I’ll break down how ReAct works and why it’s so integral to autonomous agents. Plus, I’ll introduce you to my open-source project where you can explore the mechanics of these agents without relying on any frameworks—giving you a hands-on opportunity to learn exactly what’s happening under the hood. Let’s dive into the world of AI autonomy and discover how you can start building your own intelligent agents from scratch!

The world of artificial intelligence (AI) is advancing at an unprecedented pace, with new techniques and frameworks continually pushing the boundaries of what’s possible. One of the most fascinating developments in AI agents is the use of the ReAct framework—a powerful technique that integrates reasoning and acting within language models. This approach has gained traction as the underlying methodology behind popular AI agent frameworks like AutoGPT, CrewAI, and LangChain.

What is ReAct?

At its core, ReAct stands for Reasoning and Acting, a technique designed to allow AI models to perform complex tasks autonomously. Traditionally, language models excel at processing text, but they lack the ability to act on the reasoning they produce in dynamic environments. ReAct bridges this gap by synergizing both the cognitive (reasoning) and practical (acting) sides of a task.

The ReAct technique operates by first having the AI reason about a problem—whether it’s interpreting data, formulating hypotheses, or making decisions—before moving on to act based on that reasoning. This loop of reasoning and acting allows the AI agent to handle tasks that require both deep thought and practical action, such as:

Problem-solving in real time
Multi-step task execution
Adaptation to new information and changing environments

ReAct in AI Agent Frameworks

Several AI agent frameworks have adopted ReAct to enhance the autonomy and efficiency of their agents. Notable examples include:

AutoGPT: Known for its ability to perform complex tasks autonomously, AutoGPT utilizes ReAct to break down user prompts into smaller, manageable steps, reasoning through each before acting accordingly.
CrewAI: CrewAI leverages ReAct to coordinate multiple agents in a team, enabling them to work collaboratively on tasks, reasoning through their own actions while interacting with others to achieve a common goal.
LangChain: This framework employs ReAct to help build more sophisticated applications that require both reasoning over language data and taking action, such as database queries or API calls.

In these frameworks, ReAct is pivotal to the agent’s ability to operate independently, handle dynamic tasks, and adapt as new challenges arise.

Building AI Agents Without a Framework: Learning from the Ground Up

While these frameworks provide useful abstractions for developing AI agents, understanding how ReAct works at a deeper level is key to mastering AI autonomy. That’s why I’ve created an open-source project that implements an autonomous agent using the ReAct approach—without relying on any external frameworks. It’s a hands-on way to understand the magic behind AI autonomy and build your own agent from scratch. By doing so, you’ll gain the skills needed to create sophisticated, autonomous AI agents and fully grasp the potential of ReAct.

This project is a great starting point for those who want to learn how AI agents work under the hood. By building an autonomous agent from scratch, you’ll:

Gain a solid understanding of how ReAct integrates reasoning and acting in language models.
Learn how to design an agent that can reason through a task and take action based on that reasoning.
Discover how to handle multi-step tasks and dynamic environments using only fundamental building blocks.
Avoid the abstractions of pre-built frameworks, which sometimes obscure the underlying mechanisms of AI agents.

Whether you’re an AI enthusiast or a developer looking to get hands-on experience, this project will provide valuable insights into the inner workings of autonomous AI agents.

Why Understanding ReAct is Important

ReAct is more than just a tool for AI agents—it’s a fundamental shift in how we think about the role of language models in real-world applications. By empowering models to reason and act in tandem, we unlock new possibilities for automation, problem-solving, and multi-agent collaboration. For developers and researchers, understanding ReAct is crucial for pushing the boundaries of what AI agents can achieve

Designing a Secure Architecture for Distributed Systems

Alexsandro Souza — Sun, 25 Aug 2024 20:14:09 +0000

Securing distributed systems is a complex challenge due to the diversity and scale of components involved. With multiple services interacting across potentially unsecured networks, the risk of unauthorized access and data breaches increases significantly. This article explores a practical approach to securing distributed systems using an open-source project. The project demonstrates how to integrate several security mechanisms and technologies to tackle common security challenges such as authentication, authorization, and secure communication.

Understanding Security Challenges in Distributed Systems

Distributed systems involve multiple services or microservices that must communicate securely across a network. Key security challenges in such architectures include:

Secure Communication: Ensuring that data transmitted between services is encrypted and safe from eavesdropping or tampering.
Authentication: Verifying the identities of users and services to prevent unauthorized access.
Authorization: Controlling what authenticated users and services are allowed to do, based on their roles and permissions.
Policy Enforcement: Implementing fine-grained access controls and policies that govern service-to-service and user interactions.
Certificate Management: Managing digital certificates for encrypting data and establishing trust between services.

This open-source project addresses these challenges using several integrated technologies and solutions.

Project Setup and Configuration

The project begins with setting up a secure environment using shell scripts and Docker. The setup involves provisioning digital certificates and starting the necessary services to ensure all components are ready for secure communication.

Steps to Set Up the Environment:

Provisioning Certificates: The project uses a shell script (provisioning.sh) to simulate a Certificate Authority (CA) and generate the necessary certificates for the services.

   ./provisioning.sh

Launching Services: Docker Compose is used to start all services defined in the project, ensuring they are configured correctly for secure operation.

   docker-compose up

Testing Service-to-Service Communication: To validate service-to-service communication using certificates and Jwt token, the test_services.sh script is provided. This script demonstrates how different services interact securely using their assigned certificates.

Solving Security Challenges in Distributed Systems

The project integrates several key technologies to address the primary security challenges mentioned earlier. Here's how each challenge is tackled:

1. Secure Communication with Mutual TLS (mTLS)

Challenge:
In a distributed system, services must communicate securely to prevent unauthorized access and data breaches.

Solution:
The project uses Mutual TLS (mTLS) to secure communication between services. mTLS ensures that both the client and server authenticate each other using their respective certificates. This mutual authentication prevents unauthorized services from communicating with legitimate services.

Implementation: Nginx is configured as a reverse proxy to handle mTLS. It requires both client and server certificates for establishing a secure connection, ensuring that data transmitted between services remains confidential and tamper-proof.

2. Authentication with Keycloak

Challenge:
Properly authenticating users and services is critical to prevent unauthorized access.

Solution:
The project leverages Keycloak, an open-source identity and access management solution, to manage authentication. Keycloak supports multiple authentication methods, including OpenID Connect and client credentials, making it suitable for both user and service authentication.

User Authentication:
Users are authenticated using OpenID Connect. Keycloak is configured with a client (appTest-login-client) that handles user authentication flows, including login, token issuance, and callback handling.
Service Authentication:
For service-to-service authentication, the project uses a Keycloak client (client_credentials-test) configured for the client credentials grant type. This method is ideal for authenticating services without user intervention.

Authentication Flow Example:

Users navigate to the login page.
After successful login, Keycloak redirects the user to a callback page with an authorization code.
The authorization code is then exchanged for a JWT token, which is used for subsequent requests. The authn.js file in the nginx/njs directory provides a detailed implementation of this flow.

Service Authentication Example Using Client Credentials:

curl -X POST "http://localhost:9000/realms/tenantA/protocol/openid-connect/token" \
     -H "Content-Type: application/x-www-form-urlencoded" \
     -d "grant_type=client_credentials" \
     -d "client_id=client_credentials-test" \
     -d "client_secret=your-client-secret-here"

3. User Authorization with Open Policy Agent (OPA) and JWT

Challenge:
Enforcing fine-grained access controls to ensure that authenticated users and services only have access to authorized resources.

Solution:
The project utilizes a combination of Open Policy Agent (OPA) and JWT tokens to enforce authorization policies. The project demostrate three different strategies for JWT validation to ensure robust security:

Retrieving Certificates from Keycloak: Fetches the certificates dynamically from Keycloak to validate the token.
Using x5t (Thumbprint): Uses the thumbprint embedded in the token to retrieve the public key from a local trust store.
Embedded Certificate Validation: Validates the token using an embedded certificate, ensuring the certificate is validated against a trusted Certificate Authority (CA).

Refer to the nginx/njs/token.js file for the detailed implementation of these strategies.

4. Policy Enforcement with Open Policy Agent (OPA)

Challenge:
Implementing dynamic and flexible access control policies for both services and users.

Solution:
OPA is used to enforce fine-grained policies for access control. Policies are written in a declarative language (Rego) and stored in the opa/ directory. These policies dictate the conditions under which services can communicate and users can access resources, ensuring that access controls are consistently applied across the system.

5. Certificate Management

Challenge:
Managing digital certificates for services to establish trust and secure communications.

Solution:
The project includes a robust certificate management system. A shell script (provisioning.sh) is used to simulate a Certificate Authority (CA) and generate certificates for each service. This approach simplifies certificate management and ensures that all services have the necessary credentials for secure communication.

We also added an endpoint to update the service certificate without the need of nginx restart.

curl --insecure  https://localhost/certs  --cert certificates/gen/serviceA/client.crt --key certificates/gen/serviceA/client.key -F cert=@certificates/gen/serviceA/client.crt -F key=@certificates/gen/serviceA/client.key

Conclusion

Building a secure distributed system requires careful consideration of various security aspects, including secure communication, authentication, authorization, policy enforcement, and certificate management. This open-source project provides a comprehensive example of how to integrate multiple security mechanisms to address these challenges effectively.

By following the setup and configurations demonstrated in this project, developers can leverage mutual TLS, Keycloak, Open Policy Agent, and Nginx to build a robust security architecture. These technologies, when combined, provide a strong foundation for securing distributed systems against a wide range of threats, ensuring both data protection and secure access control.

Revolutionizing User Interaction: The Rise of Conversational UIs

Alexsandro Souza — Mon, 10 Jun 2024 09:33:17 +0000

In the ever-evolving landscape of technology, the way we interact with digital interfaces is undergoing a transformative shift. Gone are the days of navigating through cumbersome menus, clicking multiple times, and dragging items across the screen to accomplish a task. The future is here, and it's conversational. Conversational User Interfaces (UIs) are set to change the game by making interactions with digital systems more natural, intuitive, and efficient. Check out a short video demo and you can try it out here

The Engine Behind Conversational UIs

At the heart of this revolutionary shift is a sophisticated engine designed to understand user intent from natural language inputs. When a user specifies what they want from the system, this engine deciphers the intent behind the given input and identifies the appropriate instructions to handle it. But how does it accomplish such a feat?

The secret sauce is the integration of Retrieval-Augmented Generation (RAG) with semantic searching, alongside the capabilities of Large Language Models (LLMs). The system leverages these technologies to not only understand the user's request but also to reason about the best way to fulfill it. Upon receiving the user's input, the engine asks the LLM to consider the retrieved documentation from the system and user input together. The LLM then responds with a structured JSON output, which includes code that can dynamically convert the structured response into a user interface complete with callbacks for submission. This process represents a leap forward in making digital interactions more human-like and less constrained by traditional UI paradigms.

Transforming Digital Navigation

Imagine wanting to input your personal information into a program. Instead of the traditional approach—navigating through a series of menus to find the right form—you simply state what information you intend to input. Instantly, the system provides you with a form pre-populated with your data, requiring only your confirmation. This scenario encapsulates the essence of conversational UIs. They promise to make our interaction with websites and applications more like having a conversation with a human assistant, where the system understands your needs from your natural language instructions.

Adding a new address to an account

Leveraging RAG and Semantic Searching

The integration of RAG with semantic searching and LLM APIs is pivotal in providing these smart solutions. RAG helps the system to pull relevant information from a vast database, while semantic searching ensures that the engine comprehends the user's intent accurately. Together, they allow the system to generate responses that are not just relevant but also contextually appropriate, paving the way for a more intelligent and responsive conversational UI.

A Practical Example: An Open Source Rich Conversational UI

To bring this concept to life, consider checking out my open-source project that exemplifies the power of rich conversational UIs. This project demonstrates how users can perform multiple tasks and display data in various formats simply through writing. Whether it's booking an appointment, filling out a survey, or querying a database, the system understands the user's request and provides the appropriate UI elements to complete the task efficiently.
This open-source example serves as a practical demonstration of how conversational UIs can transform user interaction across various domains, from customer service and e-commerce to personal assistants and beyond. By enabling a more natural way of interacting with digital systems, conversational UIs not only enhance user experience but also make technology accessible to a broader audience.

Conclusion

Conversational UIs represent a significant leap forward in the quest for more natural and intuitive user interfaces. By understanding user intent and dynamically generating appropriate UI elements, these systems promise to make our interaction with technology more like a conversation and less like a chore.

UI testing: Balancing Coverage and Cost in Software Quality Assurance

Alexsandro Souza — Wed, 03 Apr 2024 15:15:19 +0000

This post is part of a series on testing strategy, where I cover an end-to-end testing approach for web applications. Check it out!

In the quest for creating a resilient and user-friendly application, UI testing stands as a critical component of the software testing strategy. It ensures that users experience the application as intended and that the interface responds correctly to user interactions. However, an effective UI testing strategy is not just about coverage; it's about smart investment in the types of tests that provide the best return on investment. This post explores the strategic approach to UI testing, emphasizing integration testing and judicious use of end-to-end (E2E) tests.

While unit tests serve as the bedrock of a stable codebase, investing more heavily in integration tests can often yield greater dividends. Integration tests offer insights into how well different parts of the application work together, providing a closer approximation of real-world usage. For a robust testing suite, the focus should shift towards these integration tests, which effectively capture the interactions within the application.

E2E UI testing, particularly with tools like Cypress, is invaluable for verifying critical user journeys. However, these tests can be costly to run and maintain due to their complexity and the overhead involved in simulating the full range of user interactions. Given this, it’s prudent to reserve E2E UI tests for the most critical areas of your application—those user paths that are essential for the operation and revenue generation, such as checkout flows in an e-commerce app.

For integration tests, Jest emerges as an efficient and effective tool, particularly within the JavaScript and React ecosystems. Jest is well-suited for testing the interactions between components, verifying that they coalesce seamlessly to form a cohesive user experience.

In the context of a React application, where components range from simple to complex and are interdependent, integration testing verifies their combined functionality. Is important to not that the backend is not included in the integration tests, and therefore should be mocked.

Take, for instance, a to-do list application. Integration tests play a vital role in ensuring that components such as ListItem, ListItemAction, ListGroup, and TodoListView perform harmoniously. These tests would cover scenarios like adding a new task, marking a task as complete, editing an existing item, or removing it—simulating the user's perspective and actions.

In conclusion, the strategy of prioritizing integration testing over unit testing, while selectively applying E2E UI tests, aligns the development process with the ultimate goal of delivering a quality user experience. By utilizing tools like Jest for integration testing, and reserving tools like Cypress for the most critical paths, teams can maintain an effective balance between test coverage, execution cost, and maintenance overhead.

Rethinking Domain Driven Design: Beyond Tactical Patterns

Alexsandro Souza — Sat, 10 Feb 2024 15:21:21 +0000

In recent years, Domain Driven Design (DDD) has become a buzzword in the software development community, often surrounded by discussions focused on its tactical patterns: Entities, Value Objects, Repositories, Services, Aggregates, and more. This mainstream portrayal of DDD has, unfortunately, done a disservice to the software community. The emphasis on these complex patterns has led many to view DDD as overly complicated and inaccessible, particularly for projects that appear to have simpler domains. The question arises: why entangle your project in the web of DDD's tactical patterns if they seem unnecessary?

The Misguided Focus and Its Consequences

The vast majority of content and discussions available online revolve around DDD's tactical aspects, obscuring its true essence and benefits. This skewed focus has made it easy to lose sight of what DDD fundamentally aims to achieve: the strategic advantage of clearly defined boundaries within your domain, encapsulated by the concept of a Bounded Context. Unfortunately, for many newcomers to DDD, the Bounded Context—the cornerstone of DDD's strategic patterns—is not what they first encounter. Instead, they are introduced to DDD through tutorials or sample applications that dive straight into its tactical patterns without establishing the foundational understanding of the domain itself.

Avoiding Dogmatic Adherence to DDD Tactics

It's crucial for teams not to fall into what could be termed 'dogmatic DDD'—an unwavering focus on implementing DDD's tactical patterns without first grasping the strategic value of the methodology. While Entities, Aggregates, and Repositories, among other patterns, undoubtedly offer significant value, they should not precede the fundamental step of understanding and defining the domain through Bounded Contexts.

The Primacy of Bounded Contexts

A Bounded Context does more than just define the boundaries of a domain; it clarifies the domain's model and ensures that all team members share a common understanding of the domain's structure and language. This shared understanding is critical for effective communication and collaboration within the team and is often more valuable than the implementation of any specific pattern. While Bounded Contexts receive considerable attention from DDD enthusiasts, they should be the starting point for any team looking to adopt DDD, not an afterthought.

Conclusion

The essence of Domain Driven Design lies not in the complexity of its tactical patterns but in the clarity and strategic insight it brings to software architecture through the definition of Bounded Contexts. By prioritizing a deep understanding of the domain and its boundaries, teams can leverage DDD to its full potential, applying tactical patterns where beneficial but not letting them overshadow the foundational principles of the methodology. In doing so, teams can demystify DDD, making it a powerful tool for tackling complexity in software projects, regardless of the perceived simplicity or complexity of their domain. The title says it all. It’s not pattern-driven design, it’s Domain Driven Design

Practical Application: A Case Study

To illustrate these concepts, I embarked on an opensource project to design a monolithic application built on the foundations of modularity. This project serves as a practical example of how to apply the principles discussed, from domain-driven design to effective unit testing.

For a deeper dive into this project, including a walkthrough of its structure and the architectural decisions made, I invite you to check out the project and the accompanying video. This visual guide reinforces the lessons shared in this section and demonstrates how theoretical principles translate into real-world applications.

Github Project

How to create a Video chat app - Zoom(Webrtc)

Alexsandro Souza — Mon, 27 Nov 2023 10:49:14 +0000

I am always posting provoking post on my Linkedin, and I have decided to replicate those here. Let's socialize!

I have been exploring the world of WebRTC, the cutting-edge technology enabling peer-to-peer audio and video transmission. To truly understand and harness this technology, I've built a video chat(Zoom like) application from the ground up.

Key Highlights:
🔹 Signaling Server: Utilized Websockets to facilitate the exchange of WebRTC signaling messages (offer, answer, candidate) between peers. The signaling server is also responsible for broadcasting the list of active users to all connected users.

🔹 NAT Traversal: Integrated the Google STUN server in our solution to adeptly handle NAT issues, making this project not just a learning experience but a production-ready tool.

This journey through WebRTC has been incredibly interesting, and I am happy to add another great technology to my toolbox.

🔗 https://github.com/apssouza22/video-chat-rtc

I invite you all to explore, contribute, and provide feedback.

Should we use Reactive architecture with Java?

Alexsandro Souza — Thu, 23 Nov 2023 16:38:38 +0000

I am always posting provoking post on my Linkedin, and I have decided to replicate those here. Let's socialize!

Short answer:
Forget about reactive programming 😬 . In the past, it was rarely a good idea, and today, with virtual threads, it's even less recommended.
This statement from the Quarkus blog post encapsulates my point very well:

But what are virtual threads? They reuse the idea of the reactive paradigm while enabling an imperative development model. This way, you get the benefits of both reactive and imperative models without their drawbacks!

The long answer:
Reactive architecture was developed as an alternative to the high overhead of working with system threads, as Java threads utilise them. However, due to the nature of reactive architecture — one thread per core/event loop — this approach is not really recommended. As you can imagine, it's challenging to ensure there will be no blocking processes in the request. A common example of a blocking process is relational database calls (for which Quarkus offers an extension to mitigate this issue). There are other issues associated with reactive architecture, but I'll leave those for you to research.
Before virtual threads, I would have recommended reactive architecture only in very few cases (e.g., for a proxy, gateway) because once you hit certain multi thread performance limits, you could simply spawn multiple instances of the application and use a load balancer. After all, hardware is relatively cheap these days.

Now, we have virtual threads. Finally! (It has been hard to defend Java among my Golang friends) 😄
Unfortunately, there's no such thing as a free lunch, and virtual threads do have some small challenges. These challenges are described in the Quarkus post, which I won't repeat here. However, I do want to emphasise that our Platform team needs to be aware of these challenges to ensure we have a virtual thread-friendly implementations, especially around database transactions. Quarkus, Narayana and Hibernate have been patched to avoid the pinning.
Another point to note is that the platform team should ensure we have tests against thread pinning by leveraging the @ShouldNotPin annotation from junit5-virtual-threads.

Neuroevolution: Genetic Algo and Artificial Neural Network - Free Course

Alexsandro Souza — Sat, 04 Mar 2023 10:22:24 +0000

Neuroevolution is a powerful approach to machine learning and artificial intelligence that uses evolutionary algorithms to evolve neural networks.

Most neural networks use gradient descent rather than neuroevolution. However, around 2017 researchers at Uber stated they had found that simple structural neuroevolution algorithms were competitive with sophisticated modern industry-standard gradient-descent deep learning algorithms.

Deep Neuroevolution: Genetic Algorithms are a Competitive Alternative for Training Deep Neural Networks for Reinforcement Learning

If you want to learn more about this technique, check out this free course. I have created a course that introduces the principles of neuroevolution and the techniques used to design and implement neuroevolution algorithms.

The course covers the following topics:

Introduction to neuroevolution: basic principles and applications
Evolutionary algorithms: genetic algorithms, genetic programming, and evolutionary strategies
Neural networks: types, architectures, and training techniques
Neuroevolution algorithms: evolutionary algorithms applied to neural networks
Applications of neuroevolution: games, and optimization problems
Advanced topics: multi-objective neuroevolution, neuroevolution of recurrent neural networks, and deep neuroevolution.

In this project, we have applied GeneticEvolution to multiple games such as self-driving cars, smart caps and flappy bird.

This course is a follow-up to my other course about Artificial Neural Networks from scratch, where I show how to create an ANN from scratch without libraries. In that project, the learning process is done using backpropagation(gradient descent). In this project we will learn different approach. We will use Evolutionary Algorithm.

By following this course until the end, students will have a solid understanding of the principles of neuroevolution and the ability to design and implement neuroevolution algorithms for a variety of applications.

Check it out