PREFACE: WHY MOST AGENTS FORGET EVERYTHING
If you have ever built an AI agent and watched it confidently repeat the same mistake it made yesterday, you know the peculiar frustration this article is about. Most agents today are, in a very real sense, amnesiac. They wake up fresh every session, stumble into the same pitfalls, rediscover the same best practices, and then vanish into the void when the context window closes. It is like hiring a brilliant consultant who gets hit by a selective memory ray every morning before walking into your office.
This article is about fixing that. We are going to build an Agentic AI system that genuinely learns across sessions. It learns from its own failures and successes. It automatically searches the web for topics it encounters and does not fully understand. It stores everything it learns in a structured, inspectable, human-readable knowledge base called an LLM Wiki, a concept Andrej Karpathy proposed and that we will explore in considerable depth. And then, crucially, it proactively applies that accumulated knowledge when it starts a new session, so it gets smarter over time rather than staying perpetually naive.
This is not a toy. The system we build here is production-oriented, works with both local LLMs (via Ollama) and remote LLMs (via the OpenAI API), and follows clean architecture principles throughout. We will move through the design carefully, explaining every decision, so that by the end you understand not just what to build but why each piece exists and how it fits together.
STRUCTURE
CHAPTER ONE -- The Conceptual Foundation
CHAPTER TWO -- The LLM Client Abstraction
CHAPTER THREE -- The Session Logger
CHAPTER FOUR -- The Pitfall and Best Practice Detector
CHAPTER FIVE -- Shared Utilities
CHAPTER SIX -- The Autonomous Web Researcher
CHAPTER SEVEN -- The LLM Wiki Maintainer
CHAPTER EIGHT -- The Knowledge Loader
CHAPTER NINE -- The Agent Core
CHAPTER TEN -- Putting It All Together
CHAPTER ELEVEN -- The Wiki in Practice
CHAPTER TWELVE -- Advanced Patterns and Considerations
CHAPTER THIRTEEN -- What the System Looks Like After Thirty Sessions
APPENDIX -- Quick Setup Guide
CHAPTER ONE: THE CONCEPTUAL FOUNDATION
1.1 What "Agentic" Actually Means
The word "agentic" gets thrown around so casually these days that it has started to lose meaning. Let us be precise. An agentic AI system is one that does not just respond to a single prompt and stop. It perceives its environment, forms a goal, plans a sequence of steps to achieve that goal, executes those steps using tools, observes the results, and adjusts its plan accordingly. It is the difference between a calculator and a person who knows how to use a calculator, a spreadsheet, a phone, and a library, and decides which to use based on what the problem actually requires.
The key properties that make a system truly agentic are autonomy (it decides what to do next without being told step by step), tool use (it can call external functions, APIs, or services), multi-step reasoning (it can chain actions over time), and goal persistence (it keeps working toward an objective even when individual steps fail).
What most agentic systems lack, and what we are adding here, is the fifth property: learnability. The agent should become demonstrably better at its job the more it operates. Not through fine-tuning, which is expensive and requires significant infrastructure, but through structured experience accumulation that it can read and apply at the start of every new session.
1.2 The Memory Problem in Depth
Language models have no persistent state between calls. Every time you call an LLM API, you are talking to a stateless function. The "memory" that seems to exist in a long conversation is an illusion created by passing the entire conversation history as context on every call. This works fine within a session, but the moment the session ends, everything is gone.
Researchers and practitioners have proposed several approaches to address this. The simplest is just saving conversation logs to disk and loading them next time, but this scales terribly. After a hundred sessions, your context is enormous and mostly irrelevant noise. A more sophisticated approach is Retrieval-Augmented Generation (RAG), where you embed all past interactions into a vector database and retrieve semantically similar chunks when you need them. RAG works well for large unstructured corpora, but it has a fundamental weakness: it rediscovers knowledge from scratch on every query. It does not accumulate understanding. It just retrieves.
Andrej Karpathy's LLM Wiki proposal takes a different and, once you see it, obviously correct approach. Instead of retrieving raw source material, you compile it. The LLM reads raw experiences and documents, synthesizes them into structured wiki pages, and maintains those pages over time. When a new session starts, the agent reads the relevant wiki pages, which already contain synthesized, cross-referenced, contradiction-resolved knowledge. The compilation metaphor is exact: you do the hard interpretive work once, store the result, and query the result rather than the raw sources.
1.3 Karpathy's LLM Wiki: The Core Idea
The LLM Wiki is organized into three layers. The first layer is Raw Sources, which are the immutable original inputs: session logs, error reports, web search results, user feedback. The agent can read these but never modifies them. The second layer is the Wiki itself, a directory of Markdown files that the LLM writes and maintains. These files contain synthesized knowledge: concept pages, entity pages, pitfall catalogs, best practice guides, and cross-references between them. The third layer is the Schema, a configuration document (often called AGENTS.md or SCHEMA.md) that tells the LLM exactly how to maintain the wiki, what directories exist, what naming conventions to use, and what workflows to follow.
The three core operations on the wiki are Ingest (when new raw source material arrives, process it and update the wiki), Query (answer questions by reading from the wiki rather than raw sources), and Lint (periodically scan the wiki for contradictions, stale information, and orphaned pages, and fix them).
This design has a beautiful property: the wiki gets better over time. Each ingest operation can update existing pages with new information, flag contradictions, add cross-references, and refine summaries. The knowledge compounds. And because it is just Markdown files, a human can read, edit, and audit it at any time. There is no black box.
1.4 Our System Architecture
The system we are building has the following major components, which we will implement one by one.
The LLM Client is an abstraction layer that can talk to either a local Ollama instance or the OpenAI API, with a unified interface so the rest of the system does not care which backend is in use.
The Session Logger captures everything that happens during an agent session: every tool call, every result, every error, every user interaction, and metadata about what succeeded and what failed.
The Shared Utilities module provides helper functions used across multiple modules, avoiding code duplication and circular imports.
The Pitfall and Best Practice Detector is a post-session analysis module that reads the session log and uses the LLM to identify what went wrong, what went right, and what generalizable lessons can be extracted.
The Autonomous Web Researcher is a module that, after a session ends, identifies topics the agent encountered but did not fully understand, searches the web for information about them, and adds the results to the raw sources.
The Wiki Maintainer is the heart of the system. It takes raw sources (session logs, web search results, pitfall reports) and runs the Ingest, Query, and Lint operations to keep the wiki current and coherent.
The Knowledge Loader runs at the start of every new session and reads the relevant wiki pages, injecting their content into the agent's system prompt so it starts informed rather than ignorant.
The Agent Core ties everything together: it runs the main reasoning loop, calls tools, handles errors, and coordinates with all the other components.
The project directory structure looks like this before any sessions have run:
learning_agent/
agent.py
confidence_decay.py
knowledge_loader.py
llm_client.py
main.py
pitfall_detector.py
session_logger.py
utils.py
web_researcher.py
wiki_maintainer.py
.env.example
requirements.txt
raw_sources/
session_logs/
analysis/
web_research/
wiki/
(created automatically on first run)
After a few sessions the wiki directory fills with pages organized into subdirectories. The raw_sources directories accumulate JSON files that are the immutable ground truth of what the agent experienced. Everything in wiki/ is derived from raw_sources/ and can be regenerated if needed.
Let us build it.
CHAPTER TWO: THE LLM CLIENT ABSTRACTION
The first thing we need is a clean way to talk to LLMs without coupling our entire codebase to a specific provider. This is not just good engineering hygiene; it is practically necessary because you will want to use a fast local model for some tasks (like wiki maintenance, where latency matters less than cost) and a more capable remote model for others (like complex reasoning).
The design below uses a simple abstract base class and two concrete implementations. The interface is intentionally minimal: you send a list of messages and get a string back. Everything else is configuration.
# llm_client.py
#
# Unified LLM client supporting local (Ollama) and remote (OpenAI) backends.
# All agent components use this interface exclusively, so switching backends
# requires changing only the instantiation call, not the consuming code.
#
# Requirements:
# - Python 3.11+
# - For Ollama: install from https://ollama.com and run 'ollama serve'
# - For OpenAI: set the OPENAI_API_KEY environment variable
#
# No third-party packages are required. Only the Python standard library is used.
import os
import json
import urllib.request
import urllib.error
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Optional
@dataclass
class Message:
"""Represents a single message in a conversation turn."""
role: str # Must be one of: "system", "user", "assistant"
content: str
@dataclass
class LLMConfig:
"""Configuration parameters for an LLM backend."""
model: str
temperature: float = 0.2 # Low temperature for more deterministic outputs
max_tokens: int = 4096
timeout: int = 120 # Seconds to wait before giving up on a request
class LLMClient(ABC):
"""
Abstract base class for all LLM backends.
Concrete implementations must provide the 'complete' method.
All other agent components depend only on this interface, never on
a specific implementation, which makes backend switching trivial.
"""
def __init__(self, config: LLMConfig) -> None:
self.config = config
@abstractmethod
def complete(self, messages: list[Message]) -> str:
"""
Send a list of messages to the LLM and return the response text.
Args:
messages: Ordered list of conversation messages.
Returns:
The model's response as a plain string.
Raises:
RuntimeError: On unrecoverable communication or parsing errors.
"""
def complete_simple(self, system_prompt: str, user_prompt: str) -> str:
"""
Convenience wrapper for the common single-turn pattern.
Constructs a two-message conversation (system + user) and returns
the model's response. Use this for all single-shot LLM calls.
"""
return self.complete([
Message(role="system", content=system_prompt),
Message(role="user", content=user_prompt),
])
The abstract base class is deliberately thin. It does not know about retries, streaming, or function calling. Those concerns belong in higher-level components. What it establishes is the contract: give me messages, I give you text. Every component in our system depends only on this contract.
Now the Ollama implementation. Ollama runs open-weight models locally (Llama 3, Mistral, Gemma, Phi, and many others) and exposes a REST API on localhost. We use Python's built-in 'urllib' rather than the 'requests' library to keep dependencies at zero.
# llm_client.py (continued)
class OllamaClient(LLMClient):
"""
LLM client for locally-running Ollama instances.
Ollama must be running and the specified model must already be pulled.
Start Ollama: ollama serve
Pull a model: ollama pull llama3.2
List local models: ollama list
"""
def __init__(
self,
config: LLMConfig,
base_url: str = "http://localhost:11434",
) -> None:
super().__init__(config)
self.base_url = base_url.rstrip("/")
self.endpoint = f"{self.base_url}/api/chat"
def complete(self, messages: list[Message]) -> str:
payload = {
"model": self.config.model,
"messages": [
{"role": m.role, "content": m.content}
for m in messages
],
"stream": False,
"options": {
"temperature": self.config.temperature,
"num_predict": self.config.max_tokens,
},
}
data = json.dumps(payload).encode("utf-8")
req = urllib.request.Request(
self.endpoint,
data=data,
headers={"Content-Type": "application/json"},
method="POST",
)
try:
with urllib.request.urlopen(req, timeout=self.config.timeout) as resp:
body = json.loads(resp.read().decode("utf-8"))
# Ollama's /api/chat response nests the content here:
return body["message"]["content"].strip()
except urllib.error.URLError as exc:
raise RuntimeError(
f"Ollama request failed. Is Ollama running at {self.base_url}? "
f"Error: {exc}"
) from exc
except KeyError as exc:
raise RuntimeError(
f"Unexpected Ollama response format. Missing key: {exc}"
) from exc
The OpenAI client follows the same pattern. It reads the API key from an environment variable rather than accepting it as a constructor argument, which is the correct approach for secrets in any production-oriented codebase. The 'base_url' parameter makes this client compatible with any OpenAI-protocol-compatible endpoint, including Azure OpenAI, local vLLM servers, and LM Studio.
# llm_client.py (continued)
class OpenAIClient(LLMClient):
"""
LLM client for the OpenAI API and compatible endpoints.
Compatible with:
- OpenAI (api.openai.com)
- Azure OpenAI (pass your Azure endpoint as base_url)
- Local vLLM server (e.g., http://localhost:8000/v1)
- LM Studio (e.g., http://localhost:1234/v1)
Requires the OPENAI_API_KEY environment variable to be set.
For Azure, set OPENAI_API_KEY to your Azure API key.
"""
DEFAULT_ENDPOINT = "https://api.openai.com/v1/chat/completions"
def __init__(
self,
config: LLMConfig,
base_url: Optional[str] = None,
) -> None:
super().__init__(config)
# Allow a full URL override for compatible endpoints.
if base_url:
self.endpoint = base_url.rstrip("/") + "/chat/completions"
else:
self.endpoint = self.DEFAULT_ENDPOINT
self.api_key = os.environ.get("OPENAI_API_KEY", "")
if not self.api_key:
raise EnvironmentError(
"OPENAI_API_KEY environment variable is not set.\n"
"Export it before running: export OPENAI_API_KEY=sk-..."
)
def complete(self, messages: list[Message]) -> str:
payload = {
"model": self.config.model,
"messages": [
{"role": m.role, "content": m.content}
for m in messages
],
"temperature": self.config.temperature,
"max_tokens": self.config.max_tokens,
}
data = json.dumps(payload).encode("utf-8")
req = urllib.request.Request(
self.endpoint,
data=data,
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {self.api_key}",
},
method="POST",
)
try:
with urllib.request.urlopen(req, timeout=self.config.timeout) as resp:
body = json.loads(resp.read().decode("utf-8"))
return body["choices"][0]["message"]["content"].strip()
except urllib.error.HTTPError as exc:
error_body = exc.read().decode("utf-8", errors="replace")
raise RuntimeError(
f"OpenAI API error {exc.code}: {error_body}"
) from exc
except urllib.error.URLError as exc:
raise RuntimeError(
f"Network error reaching OpenAI API: {exc}"
) from exc
except (KeyError, IndexError) as exc:
raise RuntimeError(
f"Unexpected OpenAI response format. Missing key: {exc}"
) from exc
def create_llm_client(
backend: str = "ollama",
model: Optional[str] = None,
temperature: float = 0.2,
max_tokens: int = 4096,
base_url: Optional[str] = None,
) -> LLMClient:
"""
Factory function. Returns the correct LLMClient for the given backend.
Args:
backend: 'ollama' or 'openai'
model: Model name. If None, a sensible default is chosen.
temperature: Sampling temperature (0.0 = deterministic, 1.0 = creative).
max_tokens: Maximum tokens in the response.
base_url: Optional URL override for compatible endpoints.
Returns:
A configured LLMClient instance ready for use.
Raises:
ValueError: If backend is not recognized.
"""
defaults: dict[str, str] = {
"ollama": "llama3.2",
"openai": "gpt-4o-mini",
}
chosen_model = model or defaults.get(backend, "llama3.2")
config = LLMConfig(
model=chosen_model,
temperature=temperature,
max_tokens=max_tokens,
)
if backend == "ollama":
return OllamaClient(config, base_url=base_url or "http://localhost:11434")
if backend == "openai":
return OpenAIClient(config, base_url=base_url)
raise ValueError(
f"Unknown backend '{backend}'. Choose 'ollama' or 'openai'."
)
With this in place, every other component in our system can be written against the LLMClient interface. If you start with Ollama for development and switch to OpenAI for production, you change one line in your configuration file and nothing else.
CHAPTER THREE: THE SESSION LOGGER
Before the agent can learn from its experiences, it needs to record them faithfully. The session logger is the system's memory of what actually happened, as opposed to what the agent thought happened or what it reported to the user. This distinction matters enormously. Agents have a well-documented tendency to be overconfident about their own performance, so we capture objective facts: what tools were called, what arguments were passed, what was returned, how long it took, and whether an exception was raised.
The session log is stored as a JSON file, one per session, in the directory 'raw_sources/session_logs/'. This is the first layer of our LLM Wiki architecture. These files are never modified after a session ends; they are the ground truth record.
# session_logger.py
#
# Captures all agent activity during a session for post-session analysis.
# Logs are immutable once a session ends -- they are the ground truth record
# from which the wiki learns.
#
# Usage:
# with SessionLogger(goal="my task") as logger:
# logger.log_event("user_message", "Hello")
# logger.log_tool_call("read_file", {"path": "x.txt"}, "content", None, 0.1)
# logger.set_status("success")
# # Log is automatically saved when the 'with' block exits.
import json
import uuid
import time
import datetime
from dataclasses import dataclass, field, asdict
from pathlib import Path
from typing import Any, Optional
@dataclass
class ToolCall:
"""Records a single tool invocation and its outcome."""
tool_name: str
arguments: dict
result: Any # Truncated to 2000 chars before storage
error: Optional[str] # None if successful, error message if not
duration_sec: float
timestamp: str
@dataclass
class SessionEvent:
"""A generic timestamped event in the session timeline."""
event_type: str # "user_message", "agent_response", "tool_call", "error"
content: Any # Truncated to 2000 chars before storage
timestamp: str
@dataclass
class SessionLog:
"""Complete record of a single agent session."""
session_id: str
started_at: str
ended_at: Optional[str]
goal: str
tool_calls: list[ToolCall] = field(default_factory=list)
events: list[SessionEvent] = field(default_factory=list)
final_status: str = "in_progress"
notes: list[str] = field(default_factory=list)
class SessionLogger:
"""
Records all agent activity and persists the log to disk when the session ends.
Always use as a context manager so the log is saved even if the agent crashes.
The log file name encodes the date, session ID, and final status so that
directory listings are immediately informative.
"""
LOG_DIR = Path("raw_sources/session_logs")
def __init__(self, goal: str) -> None:
self.LOG_DIR.mkdir(parents=True, exist_ok=True)
self._log = SessionLog(
session_id=str(uuid.uuid4())[:8],
started_at=self._now(),
ended_at=None,
goal=goal,
)
def __enter__(self) -> "SessionLogger":
return self
def __exit__(self, exc_type, exc_val, exc_tb) -> bool:
# If we are exiting due to an unhandled exception, record it before saving.
if exc_type is not None:
self.add_note(
f"Session ended with unhandled exception: "
f"{exc_type.__name__}: {exc_val}"
)
self._log.final_status = "failure"
self.close()
return False # Do not suppress the exception; let it propagate.
def log_tool_call(
self,
tool_name: str,
arguments: dict,
result: Any,
error: Optional[str],
duration_sec: float,
) -> None:
"""
Record a tool invocation. Call this after every tool use, whether
the tool succeeded or failed.
"""
call = ToolCall(
tool_name=tool_name,
arguments=arguments,
# Truncate very large results to keep log files manageable.
result=str(result)[:2000] if result is not None else None,
error=error,
duration_sec=round(duration_sec, 3),
timestamp=self._now(),
)
self._log.tool_calls.append(call)
def log_event(self, event_type: str, content: Any) -> None:
"""Record a generic session event such as a user message or agent response."""
event = SessionEvent(
event_type=event_type,
content=str(content)[:2000],
timestamp=self._now(),
)
self._log.events.append(event)
def add_note(self, note: str) -> None:
"""Add a human-readable annotation to the session log."""
self._log.notes.append(note)
def set_status(self, status: str) -> None:
"""
Set the final session status. Call before the 'with' block exits.
Valid values: 'success', 'failure', 'partial'.
"""
self._log.final_status = status
def get_session_id(self) -> str:
"""Return the session ID for use in other components."""
return self._log.session_id
def close(self) -> Path:
"""
Finalize and persist the log to disk.
Returns:
The path to the saved log file.
"""
self._log.ended_at = self._now()
filename = (
f"{self._log.started_at[:10]}_"
f"{self._log.session_id}_"
f"{self._log.final_status}.json"
)
path = self.LOG_DIR / filename
with open(path, "w", encoding="utf-8") as f:
json.dump(asdict(self._log), f, indent=2, ensure_ascii=False)
return path
@staticmethod
def _now() -> str:
"""Return the current UTC time as an ISO 8601 string."""
return datetime.datetime.utcnow().isoformat() + "Z"
Notice that the logger truncates very large results to 2000 characters. This is a deliberate design choice. We do not want session logs to balloon to hundreds of megabytes because an agent fetched a large web page. The log captures what happened and whether it succeeded; the full content of results is not our concern here.
The context manager pattern (the 'enter' and 'exit' methods) ensures that even if the agent crashes mid-session, the log is saved. This is critical for learning from failures, which are often the most instructive events. A session that ends in a crash is not a wasted session; it is a learning opportunity, but only if the crash is recorded.
CHAPTER FOUR: THE PITFALL AND BEST PRACTICE DETECTOR
This is where things start to get genuinely interesting. After each session, we run a post-mortem analysis. We feed the session log to an LLM and ask it to think carefully about what went wrong, what went right, and what a future agent should know before starting a similar task. The output is a structured report that becomes raw material for the wiki.
The key insight here is that we are using the LLM not as an actor but as an analyst. The same model that might have made mistakes during the session is now being asked to reflect on those mistakes from the outside, reading an objective record of what happened. This works surprisingly well in practice, because the LLM is not defending its in-session decisions; it is analyzing a log file as a dispassionate observer.
# pitfall_detector.py
#
# Post-session analysis module. Reads session logs and extracts generalizable
# lessons about what went wrong, what went right, and what future agents
# should know. Output is written to raw_sources/analysis/ as JSON files.
#
# The quality of the ANALYSIS_SYSTEM_PROMPT directly determines the quality
# of the lessons extracted. Invest time in tuning it for your use case.
import json
import datetime
from pathlib import Path
from dataclasses import dataclass, field, asdict
from typing import Optional
from llm_client import LLMClient, Message
from utils import extract_json_object
@dataclass
class Lesson:
"""A single generalizable lesson extracted from a session."""
lesson_type: str # "pitfall", "best_practice", or "observation"
title: str
description: str
context: str # What situation triggers this lesson?
severity: str # "critical", "major", or "minor"
confidence: float # 0.0 to 1.0 -- how confident is the detector?
source_session: str # session_id this lesson came from
@dataclass
class AnalysisReport:
"""Complete post-session analysis for one session log."""
session_id: str
analyzed_at: str
session_goal: str
session_status: str
lessons: list[Lesson] = field(default_factory=list)
topics_to_research: list[str] = field(default_factory=list)
summary: str = ""
ANALYSIS_DIR = Path("raw_sources/analysis")
# This prompt is the most important tuning surface in the entire learning pipeline.
# The specificity and honesty of the lessons extracted depend entirely on it.
ANALYSIS_SYSTEM_PROMPT = """
You are an expert AI systems analyst specializing in agentic AI behavior.
Your job is to read session logs from an AI agent and extract generalizable
lessons that will help future agent sessions perform better.
You must respond with a valid JSON object matching this exact schema:
{
"summary": "A 2-3 sentence summary of what happened in this session.",
"lessons": [
{
"lesson_type": "pitfall" | "best_practice" | "observation",
"title": "Short title (max 10 words)",
"description": "Full explanation of the lesson (2-5 sentences)",
"context": "Describe the situation where this lesson applies",
"severity": "critical" | "major" | "minor",
"confidence": 0.0 to 1.0
}
],
"topics_to_research": [
"List of specific topics the agent encountered but clearly lacked knowledge about"
]
}
Focus on lessons that are GENERALIZABLE across sessions, not specific to this
one task. A lesson like 'always validate JSON before parsing' is generalizable.
A lesson like 'the user wanted blue widgets' is not.
Be honest about failures. Do not soften pitfalls into vague observations.
If the agent made a serious mistake, call it a critical pitfall and explain
exactly what went wrong and how to avoid it.
""".strip()
def analyze_session(
session_log_path: Path,
llm: LLMClient,
) -> AnalysisReport:
"""
Analyze a session log and return a structured report of lessons learned.
The report is also persisted to disk in raw_sources/analysis/ so that
the wiki maintainer can ingest it later.
Args:
session_log_path: Path to the JSON session log file.
llm: LLM client to use for analysis.
Returns:
An AnalysisReport containing extracted lessons and research topics.
"""
ANALYSIS_DIR.mkdir(parents=True, exist_ok=True)
with open(session_log_path, "r", encoding="utf-8") as f:
log_data = json.load(f)
session_id = log_data.get("session_id", "unknown")
session_goal = log_data.get("goal", "unknown")
session_status = log_data.get("final_status", "unknown")
# Prepare a compact but complete text representation of the session.
log_summary = _format_log_for_analysis(log_data)
messages = [
Message(role="system", content=ANALYSIS_SYSTEM_PROMPT),
Message(
role="user",
content=(
"Please analyze this agent session log and extract lessons:\n\n"
+ log_summary
),
),
]
raw_response = llm.complete(messages)
# Parse the JSON response, with graceful fallback if the LLM misbehaves.
try:
parsed = extract_json_object(raw_response)
except ValueError as exc:
# If parsing fails, create a minimal report noting the failure so
# the wiki can record that analysis itself failed -- a meta-lesson.
parsed = {
"summary": f"Analysis parsing failed: {exc}",
"lessons": [],
"topics_to_research": [],
}
lessons = [
Lesson(
lesson_type=item.get("lesson_type", "observation"),
title=item.get("title", "Untitled"),
description=item.get("description", ""),
context=item.get("context", ""),
severity=item.get("severity", "minor"),
confidence=float(item.get("confidence", 0.5)),
source_session=session_id,
)
for item in parsed.get("lessons", [])
]
report = AnalysisReport(
session_id=session_id,
analyzed_at=datetime.datetime.utcnow().isoformat() + "Z",
session_goal=session_goal,
session_status=session_status,
lessons=lessons,
topics_to_research=parsed.get("topics_to_research", []),
summary=parsed.get("summary", ""),
)
# Persist the report so the wiki maintainer can find it.
out_path = ANALYSIS_DIR / f"analysis_{session_id}.json"
with open(out_path, "w", encoding="utf-8") as f:
json.dump(asdict(report), f, indent=2, ensure_ascii=False)
return report
def _format_log_for_analysis(log_data: dict) -> str:
"""
Convert a raw session log dict into a compact, readable text representation
suitable for LLM analysis. Large result payloads are intentionally omitted
because the LLM needs the structure of what happened, not the full content.
"""
lines = [
f"SESSION ID: {log_data.get('session_id')}",
f"GOAL: {log_data.get('goal')}",
f"STATUS: {log_data.get('final_status')}",
f"STARTED: {log_data.get('started_at')}",
f"ENDED: {log_data.get('ended_at')}",
"",
"TOOL CALLS:",
]
for tc in log_data.get("tool_calls", []):
status = "ERROR" if tc.get("error") else "OK"
args_preview = json.dumps(tc.get("arguments", {}))[:200]
result_preview = str(tc.get("result", ""))[:300]
lines.append(
f" [{status}] {tc['tool_name']}({args_preview})"
f" -> {result_preview}"
f" [{tc.get('duration_sec', 0)}s]"
)
if tc.get("error"):
lines.append(f" ERROR: {tc['error']}")
lines.append("")
lines.append("EVENTS:")
for ev in log_data.get("events", []):
lines.append(
f" [{ev['event_type']}] {str(ev.get('content', ''))[:300]}"
)
if log_data.get("notes"):
lines.append("")
lines.append("NOTES:")
for note in log_data["notes"]:
lines.append(f" - {note}")
return "\n".join(lines)
The '_format_log_for_analysis' function is doing something subtle and important. It deliberately omits large result payloads and focuses on the structure of what happened: which tools were called, in what order, with what arguments, and whether they succeeded. This is the information the LLM needs to identify patterns. The full content of a file that was read, for example, is irrelevant to the question of whether the agent handled file-not-found errors correctly.
The 'topics_to_research' field in the analysis report is the trigger for our next component. When the detector identifies that the agent encountered a topic it clearly did not understand well, that topic gets queued for autonomous web research.
CHAPTER FIVE: SHARED UTILITIES
Before we continue building components, we need to address a practical engineering concern. Several modules need the same helper functions for extracting JSON from LLM output. Rather than duplicating these functions or creating circular imports by having modules import from each other, we place them in a dedicated utilities module. This is a clean architecture principle: shared low-level utilities belong in their own module with no dependencies on the rest of the system.
# utils.py
#
# Shared utility functions used across multiple agent components.
# This module has NO imports from other agent modules, which ensures
# it can be safely imported by any component without circular dependency risk.
#
# The JSON extraction functions here are necessary because LLMs, even when
# instructed to return pure JSON, frequently add preamble or postamble text.
# For example: "Here is my analysis: {...}" or "{...} I hope this helps!"
# These functions handle that gracefully.
import json
import re
def extract_json_object(text: str) -> dict:
"""
Extract the first complete JSON object from a string that may contain
surrounding text. Searches for the outermost '{' and '}' pair.
Args:
text: A string that contains a JSON object, possibly with extra text.
Returns:
The parsed JSON object as a Python dict.
Raises:
ValueError: If no JSON object is found or parsing fails.
"""
start = text.find("{")
end = text.rfind("}") + 1
if start == -1 or end == 0:
raise ValueError(
f"No JSON object found in text. First 200 chars: {text[:200]}"
)
candidate = text[start:end]
try:
return json.loads(candidate)
except json.JSONDecodeError as exc:
raise ValueError(
f"Found JSON-like text but failed to parse it: {exc}\n"
f"Candidate: {candidate[:300]}"
) from exc
def extract_json_array(text: str) -> list:
"""
Extract the first complete JSON array from a string that may contain
surrounding text. Searches for the outermost '[' and ']' pair.
Args:
text: A string that contains a JSON array, possibly with extra text.
Returns:
The parsed JSON array as a Python list.
Raises:
ValueError: If no JSON array is found or parsing fails.
"""
start = text.find("[")
end = text.rfind("]") + 1
if start == -1 or end == 0:
raise ValueError(
f"No JSON array found in text. First 200 chars: {text[:200]}"
)
candidate = text[start:end]
try:
return json.loads(candidate)
except json.JSONDecodeError as exc:
raise ValueError(
f"Found array-like text but failed to parse it: {exc}\n"
f"Candidate: {candidate[:300]}"
) from exc
def safe_filename(text: str, max_length: int = 60) -> str:
"""
Convert arbitrary text into a safe filename component.
Replaces any character that is not alphanumeric or an underscore with
an underscore, converts to lowercase, and truncates to max_length.
Args:
text: The text to convert.
max_length: Maximum number of characters in the result.
Returns:
A lowercase, underscore-separated string safe for use in filenames.
"""
cleaned = re.sub(r"[^\w]", "_", text).lower()
# Collapse multiple consecutive underscores into one.
cleaned = re.sub(r"_+", "_", cleaned).strip("_")
return cleaned[:max_length]
def truncate(text: str, max_chars: int, suffix: str = "...") -> str:
"""
Truncate text to at most max_chars characters, appending suffix if truncated.
Args:
text: The text to truncate.
max_chars: Maximum number of characters to keep.
suffix: String appended when truncation occurs.
Returns:
The original text if it fits, or a truncated version with suffix.
"""
if len(text) <= max_chars:
return text
return text[: max_chars - len(suffix)] + suffix
The 'safe_filename' function deserves a note. We use it consistently throughout the system whenever we derive a filename from user-supplied or LLM-generated text. Without it, a topic like "C++ template metaprogramming" would produce a filename with spaces and special characters that behaves differently across operating systems. The function also collapses multiple underscores, so "C++ template" becomes "c_template" rather than "c___template".
CHAPTER SIX: THE AUTONOMOUS WEB RESEARCHER
This component closes a fascinating loop. The agent encounters something it does not know well during a session. The pitfall detector notices this and flags the topic. The web researcher then automatically searches for information about that topic and stores the results as raw source material for the wiki. The next time the agent encounters the same topic, it has wiki pages synthesized from actual web research, not just its training data.
This is a form of autonomous knowledge acquisition. The agent is, in a limited but real sense, educating itself between sessions.
# web_researcher.py
#
# Autonomous web research module. Given a list of topics, searches the web
# and stores results as raw source material for wiki ingestion.
#
# Search backend: DuckDuckGo HTML interface (no API key required).
# For production use with higher volume, replace _duckduckgo_search() with
# a call to a commercial search API such as SerpAPI or Brave Search API.
#
# Output: JSON files in raw_sources/web_research/, one per researched topic.
import json
import time
import datetime
import urllib.request
import urllib.parse
import urllib.error
import html
import re
from dataclasses import dataclass, field, asdict
from pathlib import Path
from llm_client import LLMClient
from utils import safe_filename
WEB_SOURCES_DIR = Path("raw_sources/web_research")
# Politeness delay between search requests. Do not reduce this below 1 second
# or you risk being rate-limited or blocked by the search engine.
REQUEST_DELAY_SECONDS = 1.5
@dataclass
class SearchResult:
"""A single search result with title, URL, and extracted snippet."""
title: str
url: str
snippet: str
@dataclass
class ResearchReport:
"""Compiled research on a single topic, ready for wiki ingestion."""
topic: str
researched_at: str
query_used: str
results: list[SearchResult] = field(default_factory=list)
synthesis: str = "" # LLM-generated synthesis of the search results
def research_topics(
topics: list[str],
llm: LLMClient,
max_results_per_topic: int = 5,
) -> list[ResearchReport]:
"""
Research a list of topics and return compiled reports.
Each report is persisted to disk in raw_sources/web_research/ for later
wiki ingestion. A politeness delay is applied between requests.
Args:
topics: List of topic strings to research.
llm: LLM client for query crafting and synthesis.
max_results_per_topic: Maximum search results to fetch per topic.
Returns:
List of ResearchReport objects, one per topic.
"""
WEB_SOURCES_DIR.mkdir(parents=True, exist_ok=True)
reports = []
for topic in topics:
print(f" [WebResearcher] Researching: {topic}")
report = _research_single_topic(topic, llm, max_results_per_topic)
reports.append(report)
_save_report(report)
# Be polite to the search engine between requests.
time.sleep(REQUEST_DELAY_SECONDS)
return reports
def _research_single_topic(
topic: str,
llm: LLMClient,
max_results: int,
) -> ResearchReport:
"""
Research a single topic: craft a query, search, and synthesize results.
"""
query = _build_search_query(topic, llm)
results = _duckduckgo_search(query, max_results)
synthesis = ""
if results:
synthesis = _synthesize_results(topic, results, llm)
else:
synthesis = f"No search results were retrieved for topic: {topic}"
return ResearchReport(
topic=topic,
researched_at=datetime.datetime.utcnow().isoformat() + "Z",
query_used=query,
results=results,
synthesis=synthesis,
)
def _build_search_query(topic: str, llm: LLMClient) -> str:
"""
Use the LLM to craft a precise search query for the topic.
A naive query is often worse than a thoughtfully constructed one.
For example, "JSON errors" is a poor query; "Python json.JSONDecodeError
causes and handling best practices" is far more likely to return useful results.
"""
response = llm.complete_simple(
system_prompt=(
"You are a research librarian. Given a topic, produce a single, "
"precise web search query that will find the most relevant and "
"authoritative information. Return ONLY the query string, nothing else. "
"Do not add quotes, explanations, or any other text."
),
user_prompt=f"Topic: {topic}",
)
# Strip any quotes or extra whitespace the LLM might add despite instructions.
return response.strip().strip('"').strip("'").strip()
def _duckduckgo_search(query: str, max_results: int) -> list[SearchResult]:
"""
Perform a web search using DuckDuckGo's HTML interface.
This requires no API key and is suitable for moderate usage (a few searches
per session). For production use with higher volume, replace this function
with a call to a commercial search API.
Note: This function parses DuckDuckGo's HTML response with regex. If
DuckDuckGo changes their HTML structure, this parser will need updating.
"""
encoded_query = urllib.parse.quote_plus(query)
url = f"https://html.duckduckgo.com/html/?q={encoded_query}"
headers = {
"User-Agent": (
"Mozilla/5.0 (compatible; AgenticAI-Researcher/1.0; "
"educational use)"
),
}
req = urllib.request.Request(url, headers=headers)
try:
with urllib.request.urlopen(req, timeout=15) as resp:
raw_html = resp.read().decode("utf-8", errors="replace")
except urllib.error.URLError as exc:
print(f" [WebResearcher] Search request failed: {exc}")
return []
return _parse_duckduckgo_html(raw_html, max_results)
def _parse_duckduckgo_html(raw_html: str, max_results: int) -> list[SearchResult]:
"""
Extract search results from DuckDuckGo's HTML response using regex.
We use regex rather than a full HTML parser to avoid third-party dependencies.
The patterns target DuckDuckGo's result anchor elements. If zero results
are returned on a query that should have results, check whether DuckDuckGo
has changed its HTML structure and update the patterns accordingly.
"""
results = []
# Pattern for result titles and URLs.
title_pattern = re.compile(
r'class="result__a"[^>]*href="([^"]+)"[^>]*>(.*?)</a>',
re.DOTALL,
)
# Pattern for result snippets.
snippet_pattern = re.compile(
r'class="result__snippet"[^>]*>(.*?)</a>',
re.DOTALL,
)
titles = title_pattern.findall(raw_html)
snippets = snippet_pattern.findall(raw_html)
for i, (url, title_html) in enumerate(titles[:max_results]):
# Strip HTML tags and decode HTML entities from title and snippet.
title = html.unescape(
re.sub(r"<[^>]+>", "", title_html)
).strip()
raw_snippet = snippets[i] if i < len(snippets) else ""
snippet = html.unescape(
re.sub(r"<[^>]+>", "", raw_snippet)
).strip()
if title and url:
results.append(SearchResult(title=title, url=url, snippet=snippet))
return results
def _synthesize_results(
topic: str,
results: list[SearchResult],
llm: LLMClient,
) -> str:
"""
Ask the LLM to synthesize the search results into a coherent summary.
The synthesis is written for an AI agent reader: dense, specific, and
focused on practical implications rather than general background.
"""
results_text = "\n\n".join(
f"SOURCE: {r.title}\nURL: {r.url}\nSNIPPET: {r.snippet}"
for r in results
)
return llm.complete_simple(
system_prompt=(
"You are a research synthesizer. Given search results about a topic, "
"write a clear, accurate, well-organized summary (3-6 paragraphs) "
"that captures the key facts, concepts, and practical implications. "
"Write for an AI agent that needs to understand this topic to do its "
"job better. Be specific and concrete. Avoid vague generalities. "
"If the search results are low quality or irrelevant, say so explicitly."
),
user_prompt=(
f"Topic: {topic}\n\n"
f"Search Results:\n{results_text}"
),
)
def _save_report(report: ResearchReport) -> None:
"""Persist a research report to disk as a JSON file."""
filename = (
f"{report.researched_at[:10]}_"
f"{safe_filename(report.topic)}.json"
)
path = WEB_SOURCES_DIR / filename
with open(path, "w", encoding="utf-8") as f:
json.dump(asdict(report), f, indent=2, ensure_ascii=False)
One thing worth emphasizing about this module: the LLM is used twice. First, to craft a better search query than a naive string would produce. Second, to synthesize the raw search results into a coherent summary. Both uses are important. A poorly crafted query returns irrelevant results. Unsynthesized snippets are hard for the wiki maintainer to work with efficiently. The LLM acts as both a skilled researcher and a skilled writer in this pipeline.
The DuckDuckGo scraping approach is intentionally simple and transparent. In a production system, you would want to use a proper search API. The code makes this easy to swap out: just replace the '_duckduckgo_search' function with a call to your preferred API, keeping the same signature and return type.
CHAPTER SEVEN: THE LLM WIKI MAINTAINER
This is the most complex and most important component in the system. The Wiki Maintainer is responsible for keeping the LLM Wiki coherent, current, and useful. It implements Karpathy's three core operations: Ingest, Query, and Lint.
Before we look at the code, let us understand the directory structure of the wiki. This structure is itself a form of knowledge organization, and getting it right matters.
wiki/
SCHEMA.md -- The constitution of the wiki
index.md -- Master index of all pages
log.md -- Chronological record of all wiki operations
pitfalls/ -- One page per identified pitfall category
best_practices/ -- One page per best practice category
concepts/ -- General concept pages (tools, APIs, techniques)
research/ -- Synthesized research on specific topics
sessions/ -- Brief summaries of notable sessions
The SCHEMA.md file is the most important file in the entire wiki. It is the document that tells the LLM how to behave as a wiki maintainer. Without it, the LLM would produce inconsistent, poorly structured pages. With it, the wiki becomes a coherent, navigable knowledge base. Think of it as the constitution of your knowledge base: it defines the rules that all other operations must follow.
# wiki_maintainer.py
#
# The LLM Wiki Maintainer. Implements Ingest, Query, and Lint operations
# on the structured wiki knowledge base. This is the core learning engine.
#
# Based on Andrej Karpathy's LLM Wiki concept:
# https://gist.github.com/karpathy/442a6bf555914893e9891c11519de94f
#
# The wiki is a directory of plain Markdown files organized into subdirectories.
# It is human-readable, version-controllable, and directly usable by LLMs.
# No vector database or special infrastructure is required.
import json
import datetime
import re
from pathlib import Path
from typing import Optional
from llm_client import LLMClient
from utils import extract_json_array, safe_filename
WIKI_ROOT = Path("wiki")
# All wiki subdirectories. These are created automatically on initialization.
WIKI_DIRS = [
WIKI_ROOT / "pitfalls",
WIKI_ROOT / "best_practices",
WIKI_ROOT / "concepts",
WIKI_ROOT / "research",
WIKI_ROOT / "sessions",
]
# Maps lesson types from the pitfall detector to wiki subdirectories.
LESSON_TYPE_TO_DIR: dict[str, str] = {
"pitfall": "pitfalls",
"best_practice": "best_practices",
"observation": "concepts",
}
# The schema is the constitution of the wiki. It tells the LLM maintainer
# exactly how to structure and maintain pages. Changing it affects all future
# wiki operations but does not automatically update existing pages.
SCHEMA_CONTENT = """
# LLM Wiki Schema
## Purpose
This wiki is the persistent knowledge base of an agentic AI system.
It accumulates lessons from past sessions, web research, and analysis.
The agent reads this wiki at the start of each session to start informed.
## Directory Structure
- pitfalls/ Pages about failure modes, mistakes, and things to avoid.
- best_practices/ Pages about proven approaches and recommended techniques.
- concepts/ Pages explaining tools, APIs, frameworks, and concepts.
- research/ Pages synthesizing web research on specific topics.
- sessions/ Brief summaries of notable past sessions.
## Page Naming Convention
Use lowercase_with_underscores.md for all page names.
Names must be descriptive: json_parsing_errors.md, not errors.md.
## Page Structure
Every wiki page must have these sections in this order:
1. Title line: # Title
2. Metadata block: Last-Updated, Source-Sessions, Confidence (0.0-1.0)
3. Summary: 2-3 sentence overview
4. Details: The main content
5. See-Also: Links to related pages using relative paths
## Ingest Rules
- When ingesting a session analysis, create or update pages for each lesson.
- When ingesting web research, create or update pages in research/.
- Always update index.md after creating or updating pages.
- Always append an entry to log.md after any operation.
- If a new finding contradicts an existing page, note the contradiction
explicitly in the page under a Contradictions section.
- Prefer updating existing pages over creating new ones for the same topic.
## Lint Rules
- Pages with Confidence below 0.3 should be flagged for review.
- Pages not updated in 30+ days should be marked as potentially stale.
- Orphaned pages (not referenced in index.md) should be added to the index.
- Contradictions between pages should be resolved or explicitly noted.
## Quality Standards
- Be specific and concrete. Avoid vague statements.
- Every claim should trace back to a source session or research report.
- Write for an AI agent reader: be dense and precise, not conversational.
""".strip()
class WikiMaintainer:
"""
Manages the LLM Wiki: creates, updates, queries, and lints structured
knowledge pages. All wiki operations are logged to wiki/log.md.
The wiki is the single source of truth for accumulated agent knowledge.
It is designed to be human-readable and auditable at all times.
"""
def __init__(self, llm: LLMClient) -> None:
self.llm = llm
self._initialize_wiki()
def _initialize_wiki(self) -> None:
"""
Create the wiki directory structure and seed essential files.
Safe to call multiple times: existing files are never overwritten.
"""
WIKI_ROOT.mkdir(parents=True, exist_ok=True)
for directory in WIKI_DIRS:
directory.mkdir(parents=True, exist_ok=True)
schema_path = WIKI_ROOT / "SCHEMA.md"
if not schema_path.exists():
schema_path.write_text(SCHEMA_CONTENT, encoding="utf-8")
index_path = WIKI_ROOT / "index.md"
if not index_path.exists():
index_path.write_text(
"# Wiki Index\n\n"
"No pages yet. The wiki will grow as the agent learns.\n",
encoding="utf-8",
)
log_path = WIKI_ROOT / "log.md"
if not log_path.exists():
log_path.write_text("# Wiki Operation Log\n\n", encoding="utf-8")
def ingest_analysis_report(self, report_path: Path) -> None:
"""
Ingest a post-session analysis report into the wiki.
Creates or updates one wiki page per lesson extracted from the session.
Also creates a brief session summary page in wiki/sessions/.
"""
with open(report_path, "r", encoding="utf-8") as f:
report = json.load(f)
session_id = report.get("session_id", "unknown")
print(f" [WikiMaintainer] Ingesting analysis for session {session_id}")
for lesson in report.get("lessons", []):
self._ingest_lesson(lesson, session_id)
self._create_session_summary(report)
self._update_index()
self._log_operation(
"INGEST",
f"Ingested analysis for session {session_id}. "
f"{len(report.get('lessons', []))} lessons processed.",
)
def ingest_research_report(self, report_path: Path) -> None:
"""
Ingest a web research report into the wiki.
Creates or updates a page in wiki/research/ with the synthesized
findings. If a page for this topic already exists, the LLM integrates
the new information rather than overwriting the old content.
"""
with open(report_path, "r", encoding="utf-8") as f:
report = json.load(f)
topic = report.get("topic", "unknown_topic")
print(f" [WikiMaintainer] Ingesting research on: {topic}")
page_path = WIKI_ROOT / "research" / f"{safe_filename(topic)}.md"
existing_content = (
page_path.read_text(encoding="utf-8") if page_path.exists() else ""
)
sources_text = "\n".join(
f"- {r['title']}: {r['url']}"
for r in report.get("results", [])
)
prompt_context = (
f"SCHEMA:\n{SCHEMA_CONTENT}\n\n"
f"EXISTING PAGE (empty if this is a new topic):\n"
f"{existing_content}\n\n"
f"NEW RESEARCH REPORT:\n"
f"Topic: {topic}\n"
f"Researched at: {report.get('researched_at')}\n"
f"Query used: {report.get('query_used')}\n\n"
f"Synthesis:\n{report.get('synthesis', 'No synthesis available.')}\n\n"
f"Sources:\n{sources_text}"
)
new_page = self.llm.complete_simple(
system_prompt=(
"You are a wiki maintainer for an agentic AI system. "
"Given a research report and an existing wiki page (possibly empty), "
"write the complete updated wiki page that integrates the new research. "
"Follow the schema exactly. Return only the page content, no preamble."
),
user_prompt=prompt_context,
)
page_path.write_text(new_page, encoding="utf-8")
self._update_index()
self._log_operation("INGEST", f"Ingested research on topic: {topic}")
def _ingest_lesson(self, lesson: dict, session_id: str) -> None:
"""
Ingest a single lesson into the appropriate wiki subdirectory.
Reads the existing page (if any) before asking the LLM to update it.
This allows the LLM to see what is already known and either reinforce,
extend, or flag a contradiction with the existing content.
"""
lesson_type = lesson.get("lesson_type", "observation")
title = lesson.get("title", "untitled")
subdir = LESSON_TYPE_TO_DIR.get(lesson_type, "concepts")
page_path = WIKI_ROOT / subdir / f"{safe_filename(title)}.md"
existing_content = (
page_path.read_text(encoding="utf-8") if page_path.exists() else ""
)
prompt_context = (
f"SCHEMA:\n{SCHEMA_CONTENT}\n\n"
f"EXISTING PAGE (empty if this is a new topic):\n"
f"{existing_content}\n\n"
f"NEW LESSON TO INTEGRATE:\n"
f"Type: {lesson_type}\n"
f"Title: {title}\n"
f"Description: {lesson.get('description', '')}\n"
f"Context: {lesson.get('context', '')}\n"
f"Severity: {lesson.get('severity', 'minor')}\n"
f"Confidence: {lesson.get('confidence', 0.5)}\n"
f"Source Session: {session_id}\n"
)
new_page = self.llm.complete_simple(
system_prompt=(
"You are a wiki maintainer for an agentic AI system. "
"Given a lesson and an existing wiki page (possibly empty), "
"write the complete updated wiki page that integrates the new lesson. "
"If the lesson contradicts existing content, note the contradiction "
"under a Contradictions section. "
"Follow the schema exactly. Return only the page content, no preamble."
),
user_prompt=prompt_context,
)
page_path.write_text(new_page, encoding="utf-8")
def _create_session_summary(self, report: dict) -> None:
"""
Create a brief, structured summary page for a completed session.
These pages provide a chronological history of agent activity.
"""
session_id = report.get("session_id", "unknown")
page_path = WIKI_ROOT / "sessions" / f"session_{session_id}.md"
today = datetime.datetime.utcnow().isoformat()[:10]
content = (
f"# Session {session_id}\n\n"
f"Last-Updated: {today}\n"
f"Source-Sessions: {session_id}\n"
f"Confidence: 1.0\n\n"
f"## Summary\n\n"
f"{report.get('summary', 'No summary available.')}\n\n"
f"## Goal\n\n"
f"{report.get('session_goal', 'Unknown')}\n\n"
f"## Status\n\n"
f"{report.get('session_status', 'Unknown')}\n\n"
f"## Lessons Extracted\n\n"
f"{len(report.get('lessons', []))} lessons were extracted "
f"from this session.\n"
)
page_path.write_text(content, encoding="utf-8")
def query(self, question: str) -> str:
"""
Answer a question by reading from the wiki.
First identifies which pages are relevant using the index, then reads
those pages and synthesizes an answer. This is navigation, not search:
the wiki pages are already synthesized, so we read them whole.
Args:
question: The question to answer.
Returns:
A string answer citing the relevant wiki pages.
"""
index_content = (WIKI_ROOT / "index.md").read_text(encoding="utf-8")
# Ask the LLM which pages are relevant to the question.
selection_response = self.llm.complete_simple(
system_prompt=(
"You are a wiki navigator. Given a question and a wiki index, "
"identify the 3-5 most relevant page paths. "
"Return ONLY a JSON array of relative page paths, for example: "
'["pitfalls/json_errors.md", "concepts/api_calls.md"]. '
"Return an empty array if no pages are relevant."
),
user_prompt=(
f"Question: {question}\n\nWiki Index:\n{index_content}"
),
)
try:
page_paths = extract_json_array(selection_response)
except ValueError:
page_paths = []
# Read the relevant pages, respecting a character budget.
page_contents = []
total_chars = 0
char_budget = 12000 # Leave room for the question and response.
for rel_path in page_paths:
full_path = WIKI_ROOT / rel_path
if not full_path.exists():
continue
content = full_path.read_text(encoding="utf-8")
if total_chars + len(content) > char_budget:
break
page_contents.append(f"=== {rel_path} ===\n{content}")
total_chars += len(content)
if not page_contents:
return "No relevant wiki pages found for this question."
combined = "\n\n".join(page_contents)
return self.llm.complete_simple(
system_prompt=(
"You are a knowledge assistant. Answer the question using ONLY "
"the provided wiki pages. Be specific and cite which page "
"each piece of information comes from. If the pages do not "
"contain enough information to answer the question, say so."
),
user_prompt=(
f"Question: {question}\n\nWiki Pages:\n{combined}"
),
)
def lint(self) -> dict:
"""
Perform a health check on the wiki.
Finds and addresses: pages with low confidence, potentially stale pages,
and orphaned pages not listed in the index.
Returns:
A dict summarizing issues found and actions taken.
"""
print(" [WikiMaintainer] Running lint pass...")
issues: dict[str, list[str]] = {
"stale": [],
"low_confidence": [],
"orphaned": [],
"fixed": [],
}
index_content = (WIKI_ROOT / "index.md").read_text(encoding="utf-8")
now = datetime.datetime.utcnow()
for page_path in WIKI_ROOT.rglob("*.md"):
# Skip the meta-files that are maintained by the system itself.
if page_path.name in ("SCHEMA.md", "index.md", "log.md"):
continue
content = page_path.read_text(encoding="utf-8")
relative = str(page_path.relative_to(WIKI_ROOT))
# Check whether the page appears in the index.
if relative not in index_content and page_path.stem not in index_content:
issues["orphaned"].append(relative)
# Check for low confidence scores.
conf_match = re.search(r"Confidence:\s*([\d.]+)", content)
if conf_match and float(conf_match.group(1)) < 0.3:
issues["low_confidence"].append(relative)
# Check for staleness (not updated in 30+ days).
date_match = re.search(r"Last-Updated:\s*(\d{4}-\d{2}-\d{2})", content)
if date_match:
try:
last_updated = datetime.datetime.strptime(
date_match.group(1), "%Y-%m-%d"
)
if (now - last_updated).days > 30:
issues["stale"].append(relative)
except ValueError:
pass # Malformed date; ignore for now.
self._resolve_lint_issues(issues)
self._update_index()
self._log_operation(
"LINT",
f"Lint complete. Found: {len(issues['stale'])} stale, "
f"{len(issues['low_confidence'])} low-confidence, "
f"{len(issues['orphaned'])} orphaned pages. "
f"Fixed: {len(issues['fixed'])} issues.",
)
return issues
def detect_contradictions_for_page(self, page_path: Path) -> list[str]:
"""
Check a newly updated page against its related pages for contradictions.
Reads the See-Also section of the page to find related pages, then
asks the LLM to compare the pages for contradictory factual claims.
Args:
page_path: Path to the wiki page to check.
Returns:
A list of contradiction description strings (empty if none found).
"""
new_content = page_path.read_text(encoding="utf-8")
# Extract related page paths from the See-Also section.
see_also_match = re.search(
r"## See.Also\n(.*?)(?=\n##|\Z)", new_content, re.DOTALL | re.IGNORECASE
)
if not see_also_match:
return []
related_paths = re.findall(r"\(([^)]+\.md)\)", see_also_match.group(1))
contradictions: list[str] = []
for rel_path in related_paths:
related_full = WIKI_ROOT / rel_path
if not related_full.exists():
continue
related_content = related_full.read_text(encoding="utf-8")
check_result = self.llm.complete_simple(
system_prompt=(
"You are a fact-checker for a knowledge wiki. "
"Compare two wiki pages and identify any direct contradictions "
"in their factual claims. A contradiction is when Page A says X "
"and Page B says not-X about the same topic. "
"Return a JSON array of contradiction description strings, "
"or an empty array [] if no contradictions are found."
),
user_prompt=(
f"PAGE A ({page_path.name}):\n{new_content[:2000]}\n\n"
f"PAGE B ({rel_path}):\n{related_content[:2000]}"
),
)
try:
found = extract_json_array(check_result)
contradictions.extend(str(c) for c in found)
except ValueError:
pass # If parsing fails, skip this pair.
return contradictions
def _resolve_lint_issues(self, issues: dict) -> None:
"""
Apply fixes for the issues found during a lint pass.
Orphaned pages are handled by _update_index() which scans all pages.
Low-confidence pages get a review flag prepended.
Stale pages get a staleness warning prepended.
"""
for rel_path in issues["low_confidence"]:
full_path = WIKI_ROOT / rel_path
if not full_path.exists():
continue
content = full_path.read_text(encoding="utf-8")
if "REVIEW NEEDED" not in content:
flagged = "<!-- REVIEW NEEDED: Low confidence score -->\n" + content
full_path.write_text(flagged, encoding="utf-8")
issues["fixed"].append(f"Flagged for review: {rel_path}")
for rel_path in issues["stale"]:
full_path = WIKI_ROOT / rel_path
if not full_path.exists():
continue
content = full_path.read_text(encoding="utf-8")
if "POTENTIALLY STALE" not in content:
flagged = "<!-- POTENTIALLY STALE: Not updated in 30+ days -->\n" + content
full_path.write_text(flagged, encoding="utf-8")
issues["fixed"].append(f"Flagged as stale: {rel_path}")
# Orphaned pages are automatically fixed by _update_index() which
# scans all .md files regardless of their current index status.
for rel_path in issues["orphaned"]:
issues["fixed"].append(f"Will be added to index: {rel_path}")
def _update_index(self) -> None:
"""
Rebuild the wiki index by scanning all pages and collecting their titles.
The index is rebuilt from scratch on every call to ensure it is always
accurate. This is safe because the index is derived data, not source data.
"""
all_entries: list[str] = []
for page_path in sorted(WIKI_ROOT.rglob("*.md")):
if page_path.name in ("SCHEMA.md", "index.md", "log.md"):
continue
relative = str(page_path.relative_to(WIKI_ROOT))
first_line = page_path.read_text(encoding="utf-8").split("\n")[0]
title = first_line.lstrip("# ").strip() or relative
all_entries.append(f"- [{title}]({relative})")
today = datetime.datetime.utcnow().isoformat()[:10]
index_content = (
f"# Wiki Index\n\n"
f"Last-Updated: {today}\n"
f"Total pages: {len(all_entries)}\n\n"
+ "\n".join(all_entries)
+ "\n"
)
(WIKI_ROOT / "index.md").write_text(index_content, encoding="utf-8")
def _log_operation(self, operation: str, detail: str) -> None:
"""Append a timestamped entry to the wiki operation log."""
log_path = WIKI_ROOT / "log.md"
timestamp = datetime.datetime.utcnow().isoformat()[:19] + "Z"
entry = f"\n## [{timestamp}] {operation}\n{detail}\n"
with open(log_path, "a", encoding="utf-8") as f:
f.write(entry)
The 'query' method is worth examining carefully. It does not do a vector search. It does not embed the question and find similar chunks. Instead, it reads the index, asks the LLM to identify which pages are relevant, reads those pages in full, and then synthesizes an answer. This is possible because the wiki pages are already synthesized and structured. They are dense with relevant information. You do not need to retrieve fragments; you read the whole page because the page is already a coherent, focused document.
This is the fundamental advantage of the wiki approach over RAG for this use case. The knowledge has already been compiled. Retrieval is navigation, not search.
CHAPTER EIGHT: THE KNOWLEDGE LOADER
The knowledge loader is what transforms the wiki from a passive archive into an active participant in every new session. It runs at the start of each session, reads the most relevant wiki pages, and injects their content into the agent's system prompt. The agent does not have to be told to consult the wiki; the knowledge is simply there, in its context, from the moment it starts thinking.
# knowledge_loader.py
#
# Loads relevant wiki knowledge into the agent's context at session start.
# This is how accumulated learning becomes proactive behavior: the agent
# starts each session already knowing what past sessions learned.
#
# The loader enforces a character budget to avoid crowding out task context.
# Adjust MAX_WIKI_CONTEXT_CHARS based on your model's context window size.
import json
import re
from pathlib import Path
from llm_client import LLMClient
from utils import extract_json_array
WIKI_ROOT = Path("wiki")
# Maximum characters of wiki content to inject into the system prompt.
# 8000 chars is conservative. With 128K context models, you can safely
# increase this to 32000 or more.
MAX_WIKI_CONTEXT_CHARS = 8000
# Number of most-recently-modified pages to always include from each
# critical directory, regardless of the LLM's relevance selection.
CRITICAL_PAGES_PER_DIR = 2
def load_knowledge_for_session(
session_goal: str,
llm: LLMClient,
) -> str:
"""
Load the most relevant wiki knowledge for a new session.
Combines LLM-selected relevant pages with always-included critical pages
(recent pitfalls and best practices), respects a character budget, and
returns a formatted string ready to be prepended to the system prompt.
Args:
session_goal: The goal the agent is about to pursue.
llm: LLM client for relevance selection.
Returns:
A formatted knowledge context string, or an empty string if the
wiki is empty or unavailable.
"""
index_path = WIKI_ROOT / "index.md"
if not index_path.exists():
return ""
index_content = index_path.read_text(encoding="utf-8")
if "No pages yet" in index_content or len(index_content.strip()) < 50:
return ""
# Ask the LLM to select the most relevant pages for this session's goal.
selection_response = llm.complete_simple(
system_prompt=(
"You are a knowledge curator for an AI agent. "
"Given a session goal and a wiki index, select the 5 most relevant "
"wiki pages. Be selective: only include pages that are genuinely "
"relevant to the goal, not tangentially related ones. "
"Return ONLY a JSON array of relative page paths, for example: "
'["pitfalls/json_errors.md", "concepts/api_calls.md"]. '
"Return an empty array if no pages are relevant."
),
user_prompt=(
f"Session goal: {session_goal}\n\nWiki index:\n{index_content}"
),
)
try:
selected_paths = extract_json_array(selection_response)
except ValueError:
selected_paths = []
# Always include recent pitfalls and best practices regardless of selection.
all_paths = _merge_with_critical_pages(selected_paths)
# Read and assemble the selected pages within the character budget.
assembled = _assemble_pages(all_paths)
if not assembled:
return ""
return (
"=== ACCUMULATED KNOWLEDGE FROM PREVIOUS SESSIONS ===\n"
"The following knowledge was learned from past sessions and research.\n"
"Apply it proactively. Do not repeat known mistakes.\n\n"
+ assembled
+ "\n=== END OF ACCUMULATED KNOWLEDGE ===\n"
)
def _merge_with_critical_pages(selected_paths: list) -> list:
"""
Merge LLM-selected pages with always-included critical pages.
Critical pages are the most recently modified pages from the pitfalls/
and best_practices/ directories. These are always included because the
agent is most likely to repeat recent mistakes.
Args:
selected_paths: Page paths selected by the LLM for relevance.
Returns:
Combined list with duplicates removed, critical pages appended.
"""
existing = set(selected_paths)
result = list(selected_paths)
for subdir_name in ("pitfalls", "best_practices"):
subdir = WIKI_ROOT / subdir_name
if not subdir.exists():
continue
# Sort by modification time, most recent first.
pages = sorted(
subdir.glob("*.md"),
key=lambda p: p.stat().st_mtime,
reverse=True,
)
for page in pages[:CRITICAL_PAGES_PER_DIR]:
rel = str(page.relative_to(WIKI_ROOT))
if rel not in existing:
result.append(rel)
existing.add(rel)
return result
def _assemble_pages(page_paths: list) -> str:
"""
Read wiki pages and assemble them into a single context string.
Respects MAX_WIKI_CONTEXT_CHARS. Strips HTML comments (review flags)
from the content before injection so they do not confuse the agent.
Args:
page_paths: List of relative page paths to read.
Returns:
Assembled page content as a single string.
"""
parts = []
total_chars = 0
for rel_path in page_paths:
full_path = WIKI_ROOT / rel_path
if not full_path.exists():
continue
content = full_path.read_text(encoding="utf-8")
# Strip HTML comments (review flags added by the lint pass).
content = re.sub(r"<!--.*?-->", "", content, flags=re.DOTALL).strip()
if not content:
continue
if total_chars + len(content) > MAX_WIKI_CONTEXT_CHARS:
remaining = MAX_WIKI_CONTEXT_CHARS - total_chars
if remaining > 200:
# Include a truncated version rather than nothing.
parts.append(
f"--- {rel_path} ---\n"
+ content[:remaining]
+ "\n[... truncated to fit context budget ...]"
)
break
parts.append(f"--- {rel_path} ---\n{content}")
total_chars += len(content)
return "\n\n".join(parts)
The 'MAX_WIKI_CONTEXT_CHARS' limit is a practical necessity. LLM context windows are large but not infinite, and you need to leave room for the actual task. The 8000-character limit is conservative; with modern models supporting 128K or more context tokens, you could safely increase this to 32000 or beyond. The important principle is that you set a limit and enforce it, rather than letting the wiki context grow unboundedly as the wiki accumulates more pages.
The '_merge_with_critical_pages' function embodies a key design principle: some knowledge is so important that it should always be present, regardless of what the LLM thinks is relevant. Recent pitfalls and best practices are always worth including, because the agent is most likely to repeat recent mistakes.
CHAPTER NINE: THE AGENT CORE
Now we assemble everything into a working agent. The agent core implements a ReAct-style reasoning loop (Reason, Act, Observe, repeat) with tool support, session logging, and integration with all the components we have built. After the main task loop finishes, it triggers the post-session learning pipeline automatically.
The tool registry is defined first, because the agent needs it before it can do anything useful.
# agent.py
#
# The main agentic AI system. Integrates all components into a working agent
# that learns from every session and applies accumulated knowledge proactively.
#
# Architecture: ReAct reasoning loop (Reason, Act, Observe, repeat).
# The agent alternates between asking the LLM what to do next and executing
# the chosen tool, feeding the result back into the conversation.
#
# After each session, the learning pipeline runs automatically:
# 1. Analyze the session log for pitfalls and best practices.
# 2. Research any identified knowledge gaps via web search.
# 3. Ingest all findings into the LLM Wiki.
# 4. Run a lint pass to keep the wiki healthy.
import json
import re
import time
import datetime
from pathlib import Path
from dataclasses import dataclass
from typing import Callable, Any, Optional
from llm_client import LLMClient, Message, create_llm_client
from session_logger import SessionLogger
from pitfall_detector import analyze_session, ANALYSIS_DIR
from web_researcher import research_topics, WEB_SOURCES_DIR
from wiki_maintainer import WikiMaintainer
from knowledge_loader import load_knowledge_for_session
from utils import safe_filename
# Maximum number of reasoning steps before the agent gives up.
# Increase for complex multi-step tasks; decrease to save API costs during testing.
MAX_STEPS = 15
@dataclass
class Tool:
"""
A callable tool available to the agent.
The 'parameters' field is a simplified JSON Schema dict that describes
the tool's arguments. It is included in the system prompt so the LLM
knows what arguments each tool expects.
"""
name: str
description: str
parameters: dict
function: Callable
class ToolRegistry:
"""
Manages the set of tools available to the agent.
Tools are registered by name and retrieved by name during execution.
The registry also generates the tool descriptions injected into the
system prompt so the LLM knows what tools are available.
"""
def __init__(self) -> None:
self._tools: dict[str, Tool] = {}
def register(self, tool: Tool) -> None:
"""Register a tool. Overwrites any existing tool with the same name."""
self._tools[tool.name] = tool
def get(self, name: str) -> Optional[Tool]:
"""Return the tool with the given name, or None if not found."""
return self._tools.get(name)
def names(self) -> list[str]:
"""Return the names of all registered tools."""
return list(self._tools.keys())
def descriptions_for_prompt(self) -> str:
"""
Format all tool descriptions for inclusion in the system prompt.
The format is designed to be unambiguous for the LLM: tool name,
description, and parameter schema are all clearly labeled.
"""
lines = ["Available tools (use exactly these names):"]
for tool in self._tools.values():
lines.append(
f"\n {tool.name}: {tool.description}\n"
f" Parameters: {json.dumps(tool.parameters, indent=2)}"
)
return "\n".join(lines)
With the registry defined, we implement the built-in tools. These are the basic capabilities every agent needs: reading and writing files, listing directories, and doing arithmetic. In a real deployment you would add domain-specific tools here, such as database queries, API calls, or code execution.
# agent.py (continued)
def _tool_read_file(path: str) -> str:
"""Read a file from disk and return its contents as a string."""
p = Path(path)
if not p.exists():
raise FileNotFoundError(f"File not found: {path}")
if not p.is_file():
raise ValueError(f"Path is not a file: {path}")
return p.read_text(encoding="utf-8")
def _tool_write_file(path: str, content: str) -> str:
"""Write content to a file, creating parent directories as needed."""
p = Path(path)
p.parent.mkdir(parents=True, exist_ok=True)
p.write_text(content, encoding="utf-8")
return f"Successfully wrote {len(content)} characters to {path}"
def _tool_list_directory(path: str) -> str:
"""List the contents of a directory, showing type (file or dir) for each entry."""
p = Path(path)
if not p.exists():
raise FileNotFoundError(f"Directory not found: {path}")
if not p.is_dir():
raise ValueError(f"Path is not a directory: {path}")
entries = sorted(p.iterdir())
if not entries:
return f"Directory '{path}' is empty."
return "\n".join(
f"{'[DIR] ' if e.is_dir() else '[FILE]'} {e.name}"
for e in entries
)
def _tool_calculate(expression: str) -> str:
"""
Safely evaluate a mathematical expression and return the result.
Only allows digits, basic operators, parentheses, dots, commas,
percent signs, and spaces. This prevents code injection via the
expression string. For more complex math, add a dedicated library.
"""
allowed_chars = set("0123456789+-*/().,% ")
if not all(c in allowed_chars for c in expression):
disallowed = [c for c in expression if c not in allowed_chars]
raise ValueError(
f"Expression contains disallowed characters: {disallowed}. "
f"Only basic arithmetic is supported."
)
# eval with empty builtins prevents access to Python built-in functions.
result = eval(expression, {"__builtins__": {}}) # noqa: S307
return str(result)
def build_default_registry() -> ToolRegistry:
"""
Create a ToolRegistry pre-populated with the standard built-in tools.
Add your domain-specific tools here by calling registry.register()
with additional Tool instances after this function returns.
"""
registry = ToolRegistry()
registry.register(Tool(
name="read_file",
description="Read the contents of a file from disk.",
parameters={
"path": {
"type": "string",
"description": "Absolute or relative file path to read.",
}
},
function=_tool_read_file,
))
registry.register(Tool(
name="write_file",
description="Write content to a file (creates parent directories as needed).",
parameters={
"path": {
"type": "string",
"description": "Absolute or relative file path to write.",
},
"content": {
"type": "string",
"description": "The text content to write to the file.",
},
},
function=_tool_write_file,
))
registry.register(Tool(
name="list_directory",
description="List the contents of a directory.",
parameters={
"path": {
"type": "string",
"description": "Absolute or relative directory path to list.",
}
},
function=_tool_list_directory,
))
registry.register(Tool(
name="calculate",
description="Evaluate a basic mathematical expression.",
parameters={
"expression": {
"type": "string",
"description": "Math expression using +, -, *, /, (, ), e.g. '(2 + 3) * 10'",
}
},
function=_tool_calculate,
))
return registry
Now the heart of the agent: the reasoning loop. This implements the ReAct pattern, where the agent alternates between reasoning (thinking about what to do) and acting (calling a tool), observing the result, and reasoning again. The system prompt template is defined here as well, because it is intimately tied to the agent's behavior.
# agent.py (continued)
# The system prompt template. Placeholders are filled in at runtime:
# {knowledge} -- wiki content loaded by the knowledge loader
# {tools} -- tool descriptions from the registry
# {max_steps} -- the step limit, so the agent knows when to stop
AGENT_SYSTEM_PROMPT_TEMPLATE = """
You are a capable, careful AI agent. You solve tasks step by step using the
available tools. You learn from past experience and apply that knowledge proactively.
{knowledge}
{tools}
RESPONSE FORMAT:
You must respond in one of exactly two ways:
To call a tool, respond with a JSON object on a single line:
{{"action": "tool_name", "arguments": {{"param1": "value1"}}}}
To give your final answer, respond with:
{{"action": "final_answer", "answer": "Your complete answer here"}}
RULES:
- Think carefully before each action. Consider what you know from past sessions.
- If you recognize a situation that matches a known pitfall, avoid it proactively.
- Always validate inputs before passing them to tools.
- If a tool fails, analyze the error and try a different approach.
- Do not loop more than {max_steps} times. If stuck, give a partial answer.
- Never call a tool that is not in the Available tools list above.
""".strip()
class LearningAgent:
"""
An agentic AI that learns from every session it runs.
Runs a ReAct reasoning loop, logs all activity to a session log,
and triggers the full learning pipeline after each session completes.
The learning pipeline analyzes the session, researches knowledge gaps,
and updates the LLM Wiki so the next session starts smarter.
"""
def __init__(
self,
llm: LLMClient,
wiki_llm: Optional[LLMClient] = None,
registry: Optional[ToolRegistry] = None,
) -> None:
"""
Initialize the learning agent.
Args:
llm: LLM for the main agent reasoning loop.
wiki_llm: LLM for wiki maintenance tasks. Can be a smaller,
cheaper model since wiki tasks are less demanding.
Defaults to the same as 'llm' if not provided.
registry: Tool registry. Defaults to the built-in tool set.
"""
self.llm = llm
self.wiki_llm = wiki_llm or llm
self.registry = registry or build_default_registry()
self.wiki = WikiMaintainer(self.wiki_llm)
def run(self, goal: str) -> str:
"""
Run the agent on a goal and return the final answer.
Automatically triggers the post-session learning pipeline after
the session completes, whether it succeeded or failed.
Args:
goal: The task the agent should accomplish.
Returns:
The agent's final answer as a string.
"""
print(f"\n[Agent] Starting session.")
print(f"[Agent] Goal: {goal}")
with SessionLogger(goal=goal) as logger:
# Load accumulated knowledge from the wiki into the context.
knowledge_context = load_knowledge_for_session(goal, self.wiki_llm)
if knowledge_context:
print("[Agent] Loaded knowledge from wiki.")
logger.log_event(
"knowledge_loaded",
"Wiki knowledge injected into system prompt.",
)
else:
print("[Agent] Wiki is empty or has no relevant pages. Starting fresh.")
# Build the system prompt with knowledge and tool descriptions.
system_prompt = AGENT_SYSTEM_PROMPT_TEMPLATE.format(
knowledge=(
knowledge_context
or "(No prior knowledge available yet. "
"Knowledge will accumulate after this session.)"
),
tools=self.registry.descriptions_for_prompt(),
max_steps=MAX_STEPS,
)
messages = [
Message(role="system", content=system_prompt),
Message(role="user", content=f"Please accomplish this goal: {goal}"),
]
final_answer = self._reasoning_loop(messages, logger)
logger.set_status("success" if final_answer else "failure")
# The session log is now saved. Run the learning pipeline.
log_path = self._find_latest_log()
if log_path:
self._run_learning_pipeline(log_path)
return final_answer or "Task could not be completed within the step limit."
def _reasoning_loop(
self,
messages: list[Message],
logger: SessionLogger,
) -> Optional[str]:
"""
The main ReAct loop. Alternates between LLM reasoning and tool execution.
Returns the final answer string, or None if the loop exhausted its
step budget without reaching a final answer.
"""
for step in range(MAX_STEPS):
print(f" [Agent] Step {step + 1}/{MAX_STEPS}")
# Ask the LLM what to do next.
raw_response = self.llm.complete(messages)
logger.log_event("agent_response", raw_response[:500])
# Parse the JSON action from the response.
action = self._parse_action(raw_response)
if action is None:
# The LLM returned something unparseable. Nudge it back on track.
logger.log_event(
"parse_error",
f"Could not parse action from: {raw_response[:200]}",
)
messages.append(Message(role="assistant", content=raw_response))
messages.append(Message(
role="user",
content=(
"Your response was not valid JSON. Please respond with "
"exactly one of these formats:\n"
' {"action": "tool_name", "arguments": {"param": "value"}}\n'
' {"action": "final_answer", "answer": "your answer"}'
),
))
continue
# Check whether the agent has reached a final answer.
if action.get("action") == "final_answer":
answer = action.get("answer", "")
logger.log_event("final_answer", answer[:500])
print(f" [Agent] Final answer reached.")
return answer
# Execute the specified tool.
tool_name = action.get("action", "")
arguments = action.get("arguments", {})
observation = self._execute_tool(tool_name, arguments, logger)
# Feed the result back into the conversation.
messages.append(Message(role="assistant", content=raw_response))
messages.append(Message(
role="user",
content=f"Tool result:\n{observation}",
))
logger.add_note(
f"Reasoning loop exhausted {MAX_STEPS} steps without a final answer."
)
return None
def _parse_action(self, text: str) -> Optional[dict]:
"""
Extract a JSON action object from the LLM's response text.
The LLM may include reasoning text before the JSON object, so we
search for the outermost curly braces rather than parsing the whole string.
Returns:
The parsed action dict, or None if no valid JSON object was found.
"""
start = text.find("{")
end = text.rfind("}") + 1
if start == -1 or end == 0:
return None
try:
return json.loads(text[start:end])
except json.JSONDecodeError:
return None
def _execute_tool(
self,
tool_name: str,
arguments: dict,
logger: SessionLogger,
) -> str:
"""
Execute a named tool with the given arguments and return the result.
All executions are logged, including failures. Errors are returned
as strings (not raised) so the agent can observe them and adapt.
"""
tool = self.registry.get(tool_name)
if tool is None:
error_msg = (
f"Unknown tool: '{tool_name}'. "
f"Available tools: {self.registry.names()}"
)
logger.log_tool_call(tool_name, arguments, None, error_msg, 0.0)
return f"ERROR: {error_msg}"
start_time = time.monotonic()
try:
result = tool.function(**arguments)
duration = time.monotonic() - start_time
logger.log_tool_call(tool_name, arguments, result, None, duration)
return str(result)
except Exception as exc:
duration = time.monotonic() - start_time
error_msg = f"{type(exc).__name__}: {exc}"
logger.log_tool_call(tool_name, arguments, None, error_msg, duration)
return f"ERROR: {error_msg}"
def _find_latest_log(self) -> Optional[Path]:
"""
Find the most recently written session log file.
Returns None if no log files exist yet (first-ever run).
"""
log_dir = Path("raw_sources/session_logs")
if not log_dir.exists():
return None
logs = sorted(log_dir.glob("*.json"), key=lambda p: p.stat().st_mtime)
return logs[-1] if logs else None
def _run_learning_pipeline(self, log_path: Path) -> None:
"""
Run the complete post-session learning pipeline.
This is the core of the system's learnability. It runs automatically
after every session and consists of four steps:
1. Analyze the session log for pitfalls and best practices.
2. Research any identified knowledge gaps via web search.
3. Ingest all findings into the LLM Wiki.
4. Run a lint pass to keep the wiki healthy and consistent.
"""
print("\n[LearningPipeline] Starting post-session learning...")
# Step 1: Analyze the session log.
print(" [LearningPipeline] Analyzing session log...")
report = analyze_session(log_path, self.wiki_llm)
analysis_path = ANALYSIS_DIR / f"analysis_{report.session_id}.json"
# Step 2: Ingest the analysis report into the wiki.
print(" [LearningPipeline] Ingesting analysis into wiki...")
if analysis_path.exists():
self.wiki.ingest_analysis_report(analysis_path)
# Step 3: Research identified knowledge gaps.
if report.topics_to_research:
print(
f" [LearningPipeline] Researching "
f"{len(report.topics_to_research)} topics..."
)
research_reports = research_topics(
report.topics_to_research,
self.wiki_llm,
)
# Ingest each research report into the wiki.
for rr in research_reports:
rr_filename = (
f"{rr.researched_at[:10]}_"
f"{safe_filename(rr.topic)}.json"
)
rr_path = WEB_SOURCES_DIR / rr_filename
if rr_path.exists():
self.wiki.ingest_research_report(rr_path)
# Step 4: Lint the wiki to catch inconsistencies.
print(" [LearningPipeline] Running wiki lint pass...")
lint_results = self.wiki.lint()
print(
f" [LearningPipeline] Lint: "
f"{len(lint_results.get('fixed', []))} issues fixed."
)
print("[LearningPipeline] Learning pipeline complete.\n")
The '_run_learning_pipeline' method is where all the pieces come together. After every session, regardless of whether it succeeded or failed, the pipeline runs automatically. The agent analyzes what happened, researches what it did not know, updates the wiki, and cleans up any inconsistencies. The next session starts smarter.
Notice that the 'wiki_llm' can be a different, cheaper model than the main reasoning 'llm'. This is an important practical consideration. Wiki maintenance tasks do not require the most capable model available. A smaller, faster, cheaper model works fine for these tasks. The main reasoning loop, where the agent is actually solving problems, benefits from the most capable model you can afford.
CHAPTER TEN: PUTTING IT ALL TOGETHER
We need two more files to complete the system: the confidence decay module (which handles knowledge aging) and the main entry point. Let us start with confidence decay.
# confidence_decay.py
#
# Implements gradual confidence decay for wiki pages.
#
# Pages that are not reinforced by new evidence slowly lose confidence over time.
# This prevents the wiki from treating old, potentially stale knowledge with the
# same authority as recently validated knowledge.
#
# Call apply_decay_to_all_pages() periodically (e.g., weekly) as part of
# maintenance. It is integrated into main.py's --maintain-wiki command.
import re
import datetime
from pathlib import Path
WIKI_ROOT = Path("wiki")
# Confidence lost per day without reinforcement.
# At this rate, a page loses half its confidence in 100 days.
DECAY_RATE_PER_DAY = 0.005
# Confidence never decays below this floor, ensuring pages are always
# flagged for review rather than silently disappearing.
MIN_CONFIDENCE = 0.1
def apply_decay_to_all_pages() -> dict:
"""
Apply confidence decay to all wiki pages based on their last-updated date.
For each page, calculates how many days have passed since it was last
updated and reduces its confidence score proportionally. Pages at or
below 0.3 confidence will be flagged by the next lint pass.
Returns:
A dict with 'decayed' (list of page paths that were updated) and
'skipped' (list of pages that could not be processed).
"""
now = datetime.datetime.utcnow()
results = {"decayed": [], "skipped": []}
for page_path in WIKI_ROOT.rglob("*.md"):
if page_path.name in ("SCHEMA.md", "index.md", "log.md"):
continue
content = page_path.read_text(encoding="utf-8")
conf_match = re.search(r"(Confidence:\s*)([\d.]+)", content)
date_match = re.search(r"Last-Updated:\s*(\d{4}-\d{2}-\d{2})", content)
if not conf_match or not date_match:
results["skipped"].append(str(page_path))
continue
current_conf = float(conf_match.group(2))
try:
last_updated = datetime.datetime.strptime(
date_match.group(1), "%Y-%m-%d"
)
except ValueError:
results["skipped"].append(str(page_path))
continue
days_since_update = (now - last_updated).days
if days_since_update <= 0:
continue
# Apply linear decay clamped to the minimum floor.
decay = DECAY_RATE_PER_DAY * days_since_update
new_conf = max(MIN_CONFIDENCE, current_conf - decay)
# Only rewrite the file if the change is meaningful (> 0.001).
if abs(new_conf - current_conf) > 0.001:
updated_content = re.sub(
r"(Confidence:\s*)[\d.]+",
f"Confidence: {new_conf:.3f}",
content,
)
page_path.write_text(updated_content, encoding="utf-8")
results["decayed"].append(str(page_path))
return results
Now the main entry point, which ties everything together and provides a clean command-line interface for all system operations.
# main.py
#
# Entry point for the Learning Agent system.
#
# Usage examples:
#
# Run the agent on a task (Ollama backend):
# python main.py --backend ollama --goal "Summarize the files in ./docs"
#
# Run the agent on a task (OpenAI backend):
# python main.py --backend openai --model gpt-4o --goal "Analyze ./data"
#
# Use different models for agent and wiki (recommended for cost efficiency):
# python main.py --backend openai --model gpt-4o \
# --wiki-backend openai --wiki-model gpt-4o-mini \
# --goal "Your task here"
#
# Run a wiki lint pass without starting a new session:
# python main.py --lint-wiki
#
# Query the wiki directly:
# python main.py --query-wiki "What are the most common pitfalls?"
#
# Run confidence decay maintenance:
# python main.py --maintain-wiki
#
# Use a custom OpenAI-compatible endpoint (e.g., local vLLM):
# python main.py --backend openai --base-url http://localhost:8000/v1 \
# --model my-model --goal "Your task here"
import argparse
import sys
from pathlib import Path
from llm_client import create_llm_client, LLMClient
from agent import LearningAgent
from wiki_maintainer import WikiMaintainer
from confidence_decay import apply_decay_to_all_pages
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(
description="Learning Agentic AI System with LLM Wiki Memory",
formatter_class=argparse.RawDescriptionHelpFormatter,
)
# LLM backend configuration.
parser.add_argument(
"--backend",
choices=["ollama", "openai"],
default="ollama",
help="LLM backend for the main agent (default: ollama)",
)
parser.add_argument(
"--model",
default=None,
help=(
"Model name for the main agent. "
"Defaults: llama3.2 (ollama), gpt-4o-mini (openai)"
),
)
parser.add_argument(
"--base-url",
default=None,
dest="base_url",
help=(
"Custom base URL for OpenAI-compatible endpoints "
"(e.g., http://localhost:8000/v1 for vLLM)"
),
)
# Wiki LLM configuration (can be a different, cheaper model).
parser.add_argument(
"--wiki-backend",
choices=["ollama", "openai"],
default=None,
dest="wiki_backend",
help="LLM backend for wiki maintenance (default: same as --backend)",
)
parser.add_argument(
"--wiki-model",
default=None,
dest="wiki_model",
help="Model for wiki maintenance (can be smaller/cheaper than main model)",
)
parser.add_argument(
"--wiki-base-url",
default=None,
dest="wiki_base_url",
help="Custom base URL for the wiki LLM endpoint",
)
# Agent task.
parser.add_argument(
"--goal",
default=None,
help="The task goal for the agent to accomplish",
)
# Standalone maintenance operations.
parser.add_argument(
"--lint-wiki",
action="store_true",
dest="lint_wiki",
help="Run a wiki lint pass and exit (no agent session)",
)
parser.add_argument(
"--query-wiki",
default=None,
dest="query_wiki",
metavar="QUESTION",
help="Query the wiki and print the answer (no agent session)",
)
parser.add_argument(
"--maintain-wiki",
action="store_true",
dest="maintain_wiki",
help="Run confidence decay and lint, then exit (no agent session)",
)
return parser.parse_args()
def _build_wiki_llm(args: argparse.Namespace, main_llm: LLMClient) -> LLMClient:
"""
Build the wiki LLM client.
If wiki-specific backend/model are specified, create a separate client.
Otherwise, reuse the main LLM client to avoid redundant connections.
"""
wiki_backend = args.wiki_backend
wiki_model = args.wiki_model
wiki_base_url = args.wiki_base_url
# If no wiki-specific settings were provided, reuse the main client.
if not wiki_backend and not wiki_model and not wiki_base_url:
return main_llm
# Build a separate wiki client with the specified settings.
effective_backend = wiki_backend or args.backend
print(
f"[Main] Wiki LLM: {effective_backend} / "
f"{wiki_model or 'default'}"
)
return create_llm_client(
backend=effective_backend,
model=wiki_model,
temperature=0.1, # Lower temperature for more consistent wiki maintenance.
base_url=wiki_base_url,
)
def main() -> None:
args = parse_args()
# Build the main LLM client.
print(
f"[Main] Initializing LLM client: "
f"{args.backend} / {args.model or 'default'}"
)
main_llm = create_llm_client(
backend=args.backend,
model=args.model,
temperature=0.2,
base_url=args.base_url,
)
# Build the wiki LLM client (may be the same or a separate cheaper model).
wiki_llm = _build_wiki_llm(args, main_llm)
# Handle standalone wiki maintenance operations.
if args.maintain_wiki:
print("[Main] Running wiki maintenance (decay + lint)...")
decay_results = apply_decay_to_all_pages()
print(
f"[Main] Decay applied to {len(decay_results['decayed'])} pages. "
f"Skipped {len(decay_results['skipped'])} pages."
)
wiki = WikiMaintainer(wiki_llm)
lint_results = wiki.lint()
print(f"[Main] Lint complete: {lint_results}")
return
if args.lint_wiki:
print("[Main] Running wiki lint pass...")
wiki = WikiMaintainer(wiki_llm)
lint_results = wiki.lint()
print(f"[Main] Lint complete: {lint_results}")
return
if args.query_wiki:
print(f"[Main] Querying wiki: {args.query_wiki}")
wiki = WikiMaintainer(wiki_llm)
answer = wiki.query(args.query_wiki)
print(f"\n[Wiki Answer]\n{answer}")
return
# Run the agent on the specified goal.
if not args.goal:
print(
"Error: --goal is required when running the agent.\n"
"Example: python main.py --goal 'List the Python files in ./src'"
)
sys.exit(1)
agent = LearningAgent(llm=main_llm, wiki_llm=wiki_llm)
answer = agent.run(args.goal)
print(f"\n[Final Answer]\n{answer}")
if __name__ == "__main__":
main()
CHAPTER ELEVEN: THE WIKI IN PRACTICE
To make this concrete, let us look at what a wiki page actually looks like after a few sessions. Suppose the agent repeatedly encountered issues with JSON parsing errors in API responses. The wiki page that emerges from several sessions of experience might look like this:
# JSON Parsing Errors in API Responses
Last-Updated: 2025-07-18
Source-Sessions: a3f1b2, 9d4e7c, 2b8f1a
Confidence: 0.92
## Summary
API responses frequently contain malformed JSON, trailing commas, or
non-JSON error messages that cause json.loads() to raise ValueError.
This is one of the most common failure modes in agentic tool use.
## Details
The most reliable pattern is to always wrap json.loads() in a try/except
block and have a fallback strategy. In session a3f1b2, the agent crashed
because it assumed a 200 OK response always contained valid JSON. In
session 9d4e7c, the same mistake was made with a different API. In session
2b8f1a, the agent correctly handled the error by checking the Content-Type
header first.
Recommended approach:
1. Check the Content-Type header before attempting to parse.
2. Use try/except json.JSONDecodeError (not a bare except clause).
3. Log the raw response text when parsing fails for debugging.
4. Have a fallback: treat the raw text as the result if JSON parsing fails.
The extract_json_object() helper in utils.py is more robust than raw
json.loads() for LLM outputs specifically, because it handles surrounding
text that LLMs often add despite instructions.
## See Also
- concepts/json_extraction.md
- best_practices/api_error_handling.md
- pitfalls/assuming_successful_responses.md
This page is the result of three sessions of accumulated experience. It is specific, actionable, and cross-referenced. When the agent starts its fourth session involving API calls, this page is injected into its context, and it knows to check the Content-Type header before parsing, to use try/except correctly, and to log raw responses on failure. It will not make those mistakes again.
This is what learning looks like in practice. Not a gradient update. Not a fine-tuning run. Just structured, accumulated, human-readable experience that compounds over time.
CHAPTER TWELVE: ADVANCED PATTERNS AND CONSIDERATIONS
12.1 The Confidence Decay Problem
One subtle issue with any persistent knowledge system is that knowledge can become stale. A best practice that was correct in 2023 might be outdated in 2025. The Lint operation addresses this with the 30-day staleness check, but the confidence decay module goes further by implementing a continuous decay function. Every day that passes without a page being reinforced by new evidence, its confidence score decreases slightly. Pages that are consistently reinforced by new sessions maintain high confidence. Pages that are never reinforced gradually decay toward the review threshold, where the lint pass will flag them for human inspection.
The decay rate of 0.005 per day means a page loses about 18% of its confidence over a month and about half its confidence over 100 days. You can tune this rate based on how quickly your domain evolves. A rapidly changing field like security vulnerabilities might warrant a higher decay rate; a stable domain like mathematical algorithms might warrant a much lower one.
12.2 Contradiction Detection
When the wiki maintainer ingests a new lesson that contradicts an existing page, it should not silently overwrite the old knowledge. The 'detect_contradictions_for_page' method in WikiMaintainer handles this by reading the See-Also section of a newly updated page, finding the related pages, and asking the LLM to compare them for contradictory claims. When contradictions are found, they are noted in the page under a Contradictions section rather than being silently resolved. This preserves the history of conflicting evidence and lets a human or a future lint pass make an informed decision about which version is correct.
The method is already integrated into the WikiMaintainer class. You can call it explicitly after ingesting a lesson if you want immediate contradiction checking, or let the lint pass surface contradictions periodically.
12.3 Scaling to Multiple Agents
The architecture we have built is designed for a single agent, but it scales naturally to multi-agent systems. Multiple agents can share the same wiki directory. Each agent runs its own learning pipeline after its sessions, contributing to a shared knowledge base. The Lint operation becomes especially important in this setting, because different agents may develop contradictory beliefs based on different experiences.
The key design principle for multi-agent wikis is that the wiki is the single source of truth. Agents do not share beliefs directly; they share them through the wiki. This prevents the kind of echo-chamber dynamics that can emerge when agents communicate directly, where one agent's confident but wrong belief propagates to others without being checked against evidence. Every belief must be written to the wiki, where it is subject to the same lint and contradiction-detection processes as any other piece of knowledge.
12.4 Choosing Models for Each Role
The system supports using different models for the main agent reasoning loop and for wiki maintenance. This is worth doing in practice because the two roles have different requirements. The main reasoning loop benefits from the most capable model available, because it is solving the actual problem and its quality directly affects the user. Wiki maintenance tasks (ingesting lessons, updating pages, running lint) are more formulaic and benefit from consistency and speed rather than raw capability. A smaller, faster model like Phi-3 or GPT-4o-mini works well for wiki maintenance and costs significantly less than using a frontier model for everything.
A practical configuration for a production system might use GPT-4o for the main agent and GPT-4o-mini for the wiki, or Llama 3.2 70B for the main agent and Phi-3 for the wiki when running locally.
CHAPTER THIRTEEN: WHAT THE SYSTEM LOOKS LIKE AFTER THIRTY SESSIONS
After thirty sessions of varied tasks, the wiki might contain forty or fifty pages organized across the five directories. The pitfalls directory might have pages on JSON parsing errors, file encoding issues, tool argument validation failures, and the dangers of assuming that a 200 HTTP status code means the response is valid. The best_practices directory might have pages on validating inputs before tool calls, using try/except with specific exception types, checking file existence before reading, and structuring multi-step tasks as explicit plans before execution.
The research directory might have pages synthesizing web research on topics like Python file encoding detection, REST API error handling patterns, and directory traversal security considerations. The concepts directory might have pages on the tools the agent uses most frequently, with accumulated notes about their quirks and edge cases.
Every new session starts by reading the most relevant subset of these pages. The agent that starts session thirty-one is, in a meaningful sense, a different agent from the one that started session one. It has the accumulated wisdom of thirty sessions of experience. It knows what mistakes to avoid. It knows which approaches have worked. It knows where to look for information when it is uncertain.
This is what learnability means in practice. Not a magic property that emerges from scale, but a deliberate architectural choice: capture experience, synthesize it, store it accessibly, and inject it proactively. The LLM Wiki is the mechanism that makes this possible, and Karpathy's insight that knowledge should be compiled rather than retrieved is the key that unlocks it.
The wiki itself becomes an artifact worth reading. After thirty sessions, you will find that the pages contain insights about your specific domain, your specific tools, and your specific failure modes that no general-purpose model would know. The wiki is personalized to your workload in a way that no amount of prompt engineering can replicate, because it is grounded in actual experience rather than general training.
APPENDIX: QUICK SETUP GUIDE
A.1 Prerequisites
You need Python 3.11 or later. No third-party Python packages are required; the entire system uses only the Python standard library. For the LLM backends, you need either Ollama installed locally or an OpenAI API key.
A.2 Project File List
The complete project consists of these files, all of which are defined in full in this article:
learning_agent/
agent.py -- Agent core, tool registry, learning pipeline
confidence_decay.py -- Wiki confidence decay for knowledge aging
knowledge_loader.py -- Loads wiki knowledge into session context
llm_client.py -- LLM abstraction (Ollama + OpenAI)
main.py -- Command-line entry point
pitfall_detector.py -- Post-session analysis and lesson extraction
session_logger.py -- Session activity recorder
utils.py -- Shared utility functions
web_researcher.py -- Autonomous web research module
wiki_maintainer.py -- LLM Wiki ingest, query, and lint operations
.env.example -- Template for environment variables
requirements.txt -- Dependency declaration (stdlib only)
A.3 The requirements.txt File
# requirements.txt
#
# The Learning Agent system uses only the Python standard library.
# No third-party packages are required.
#
# Python version requirement: 3.11 or later
#
# Optional: if you want to use the openai Python SDK instead of the
# built-in urllib implementation, install it with:
# pip install openai>=1.0.0
#
# Optional: for higher-volume web search, install a search API client:
# pip install serpapi (SerpAPI)
# pip install brave-search (Brave Search API)
A.4 The .env.example File
# .env.example
#
# Copy this file to .env and fill in your values.
# Then load it before running: source .env (Linux/macOS)
# or use python-dotenv in your code.
#
# Required for OpenAI backend:
OPENAI_API_KEY=sk-your-key-here
#
# Optional: custom OpenAI-compatible endpoint
# OPENAI_BASE_URL=http://localhost:8000/v1
#
# Optional: Ollama configuration (defaults shown)
# OLLAMA_BASE_URL=http://localhost:11434
A.5 Installing and Starting Ollama
Download and install Ollama from https://ollama.com. Then pull the models you want to use and start the server:
ollama pull llama3.2
ollama pull phi3
ollama serve
The 'ollama serve' command starts the API server on port 11434. It runs in the foreground; use a separate terminal for your agent commands, or run it as a background service.
A.6 Running Your First Session
Create the project directory, place all files in it, and run:
cd learning_agent
python main.py --backend ollama --model llama3.2 \
--wiki-backend ollama --wiki-model phi3 \
--goal "List the Python files in the current directory and count their lines"
You will see output like this:
[Main] Initializing LLM client: ollama / llama3.2
[Main] Wiki LLM: ollama / phi3
[Agent] Starting session.
[Agent] Goal: List the Python files in the current directory and count their lines
[Agent] Wiki is empty or has no relevant pages. Starting fresh.
[Agent] Step 1/15
[Agent] Step 2/15
[Agent] Step 3/15
[Agent] Final answer reached.
[LearningPipeline] Starting post-session learning...
[LearningPipeline] Analyzing session log...
[LearningPipeline] Ingesting analysis into wiki...
[LearningPipeline] Researching 2 topics...
[WebResearcher] Researching: Python file line counting best practices
[WebResearcher] Researching: pathlib glob patterns for Python files
[LearningPipeline] Running wiki lint pass...
[LearningPipeline] Lint: 0 issues fixed.
[LearningPipeline] Learning pipeline complete.
[Final Answer]
Found 10 Python files. Total line count: 847 lines.
After this first session, the wiki directory exists and contains pages. Run the same command again (or a similar one) and you will see "Loaded knowledge from wiki." instead of "Starting fresh." The agent now starts with the benefit of the first session's experience.
A.7 Running with OpenAI
Set your API key and run:
export OPENAI_API_KEY=sk-your-key-here
python main.py --backend openai --model gpt-4o \
--wiki-backend openai --wiki-model gpt-4o-mini \
--goal "Your task here"
A.8 Wiki Maintenance Commands
Run a lint pass to check wiki health without starting a new session:
python main.py --lint-wiki
Run confidence decay and lint together (recommended weekly):
python main.py --maintain-wiki
Query the wiki directly to see what the agent has learned:
python main.py --query-wiki "What are the most common pitfalls this agent encounters?"
A.9 Inspecting the Wiki
The wiki is just a directory of Markdown files. You can read them with any text editor, open the directory in Obsidian for a graph view, or version-control it with Git. The log file at wiki/log.md provides a chronological record of every wiki operation. The index at wiki/index.md lists every page with a link.
After a few sessions, try reading wiki/pitfalls/ and wiki/best_practices/. You will find that the pages contain specific, actionable knowledge grounded in actual agent experience. That is the system working as intended.
A.10 Adding Custom Tools
To add a domain-specific tool, define a function and register it with the tool registry before creating the agent:
# Example: adding a custom tool to the agent
from agent import LearningAgent, ToolRegistry, Tool, build_default_registry
from llm_client import create_llm_client
def my_custom_tool(query: str) -> str:
"""Your custom tool implementation."""
# ... your logic here ...
return f"Result for: {query}"
llm = create_llm_client(backend="ollama", model="llama3.2")
registry = build_default_registry()
registry.register(Tool(
name="my_tool",
description="Does something specific to my domain.",
parameters={
"query": {
"type": "string",
"description": "The query to process.",
}
},
function=my_custom_tool,
))
agent = LearningAgent(llm=llm, registry=registry)
answer = agent.run("Use my_tool to process the quarterly report.")
print(answer)
The agent will automatically learn from its experience using your custom tool, just as it learns from the built-in tools. If the tool has quirks or failure modes, they will appear in the wiki after a few sessions.
