akgentic-llm 1.6.0

LLM integration layer for Akgentic agent systems

akgentic-llm

LLM integration layer for the Akgentic multi-agent framework. Wraps pydantic-ai's REACT execution loop with persistent context management, production HTTP retry logic, and a clean provider abstraction — letting agents call any LLM without coupling to a specific vendor or framework primitive.

Table of Contents

• Overview
• Installation
• Quick Start
• Configuration
  • ModelConfig
  • UsageLimits
  • RuntimeConfig
  • ReactAgentConfig
• Providers
• ReactAgent API
• Multimodal Prompts
• Context Management
  • ContextManager
  • Observer Pattern
  • Tool Event Observability
  • System Prompt Rendering Events
• Context Compaction
  • Compaction & Clear Events
  • Compaction Strategies
  • Overriding the Summarizer Prompt
• Cost Tracking and Aggregation
• Prompts
• Development
• License

Overview

akgentic-llm is the LLM execution layer between agent logic and LLM providers. It provides:

• ReactAgent — a thin wrapper around pydantic-ai's Agent.iter() that persists message history across calls, deduplicates messages across tool-call iterations, and translates pydantic-ai's UsageLimitExceeded into a framework-local UsageLimitError
• Provider abstraction — create_model() dispatches to one of six provider factories (OpenAI, Azure, Anthropic, Google, Mistral, NVIDIA); get_output_type() wraps output types with NativeOutput for providers that support structured output, falls back to prompt-based extraction for those that don't
• HTTP retry — create_http_client() configures AsyncTenacityTransport with exponential backoff, jitter, and Retry-After header support; fast-fails on 4xx (except 429)
• Context management — ContextManager tracks message history across multiple run() calls, supports checkpoint/rewind for error recovery, and applies a sliding window (system messages always preserved) when a message cap is configured
• Prompt utilities — PromptTemplate for config-time {placeholder} rendering; current_datetime_prompt and json_output_reminder_prompt as ready-made dynamic prompts
• Multimodal — UserPrompt = str | list[str | BinaryContent]; exported so akgentic-agent can annotate its own act() signature without importing pydantic-ai directly

ReactAgent
  │
  ├── run(user_prompt: UserPrompt)           # str | list[str | BinaryContent]
  │     │
  │     ├── pydantic_agent.iter(            # pydantic-ai REACT loop
  │     │       user_prompt,
  │     │       message_history=context.messages,
  │     │       output_type=get_output_type(model_cfg, output_type),
  │     │   )
  │     │     │
  │     │     └── for each step:
  │     │           context.add_message()   # persists + notifies observers
  │     │
  │     └── return run.result.output
  │
  ├── context: ContextManager               # persistent message history
  ├── checkpoint() / rewind()               # snapshot and restore context
  └── system_prompt(func)                   # register dynamic system prompt

Module boundary: akgentic-llm depends only on pydantic-ai, httpx, and tenacity. It MUST NOT import from akgentic-core, akgentic-tool, or akgentic-agent.

Installation

Workspace Installation (Recommended)

git clone [e-mail]:b12consulting/akgentic-quick-start.git
cd akgentic-quick-start
git submodule update --init --recursive

uv venv && source .venv/bin/activate
uv sync --all-packages --all-extras

Standalone

cd packages/akgentic-llm
uv venv && source .venv/bin/activate
uv pip install -e ".[dev]"

Quick Start

from akgentic.llm import ReactAgent, ReactAgentConfig, ModelConfig

config = ReactAgentConfig(
    model_cfg=ModelConfig(provider="openai", model="gpt-4o")
)

agent = ReactAgent(config=config)
result = agent.run_sync("Summarise the key priorities for next sprint.")
print(result)

With tools and a per-call output type:

from pydantic import BaseModel
from akgentic.llm import ReactAgent, ReactAgentConfig, ModelConfig, UsageLimits

class Summary(BaseModel):
    title: str
    points: list[str]

def fetch_data(topic: str) -> str:
    """Retrieve data about a topic."""
    return f"Latest data on {topic}: ..."

agent = ReactAgent(
    config=ReactAgentConfig(
        model_cfg=ModelConfig(provider="anthropic", model="claude-3-5-sonnet-20241022"),
        usage_limits=UsageLimits(request_limit=10, total_tokens_limit=20_000),
    ),
    tools=[fetch_data],
)

result = agent.run_sync("Summarise AI trends", output_type=Summary)
print(result.title, result.points)

Configuration

ModelConfig

Field	Type	Default	Description
provider	Literal[...]	"openai"	LLM provider
model	str	"gpt-5.2"	Model identifier (provider-specific)
temperature	float | None	None	0.0–2.0; None = provider default
seed	int | None	None	Reproducible outputs (not all providers)
max_tokens	int | None	None	Max response tokens; None = provider max
reasoning_effort	Literal["low","medium","high"] | None	None	For o1/o3-style models only

from akgentic.llm import ModelConfig

# Standard chat model
ModelConfig(provider="openai", model="gpt-4o", temperature=0.7)

# Deterministic with token cap
ModelConfig(provider="anthropic", model="claude-3-5-sonnet-20241022",
            temperature=0.0, seed=42, max_tokens=2000)

# Reasoning model
ModelConfig(provider="openai", model="o1", reasoning_effort="high")

UsageLimits

Limits are cumulative across all requests in a single run() call. Breaching any limit raises UsageLimitError.

Field	Type	Default	Description
request_limit	int | None	50	Max LLM API requests — acts as a safety brake
tool_calls_limit	int | None	None	Max tool invocations
input_tokens_limit	int | None	None	Max cumulative input tokens
output_tokens_limit	int | None	None	Max cumulative output tokens
total_tokens_limit	int | None	None	Max cumulative total tokens

from akgentic.llm import UsageLimits

UsageLimits(request_limit=10, total_tokens_limit=5_000)   # tight budget
UsageLimits(request_limit=None)                            # unlimited (no safety brake)

RuntimeConfig

Field	Type	Default	Description
retries	int	3	Retry attempts for tool failures and output validation errors
end_strategy	Literal["early","exhaustive"]	"exhaustive"	Tool execution termination
parallel_tool_calls	bool	True	Concurrent tool execution when provider supports it
http_client_config	HttpClientConfig	HttpClientConfig()	HTTP timeout and retry tuning

End strategies:

• "early" — stops after the first successful result (fast path)
• "exhaustive" — runs all tool calls even when a result is available (complete data gathering)

Note: parallel_tool_calls is silently forced to False for providers without native structured output (google-gla, mistral, non-openai NVIDIA). See Providers.

HttpClientConfig fields: timeout=120.0, max_retries=5, backoff_multiplier=0.5, backoff_max=60.0 — all configurable.

ReactAgentConfig

Composes all three layers:

from akgentic.llm import ReactAgentConfig, ModelConfig, UsageLimits, RuntimeConfig, HttpClientConfig

config = ReactAgentConfig(
    model_cfg=ModelConfig(
        provider="anthropic",
        model="claude-3-5-sonnet-20241022",
        temperature=0.7,
    ),
    usage_limits=UsageLimits(
        request_limit=10,
        total_tokens_limit=50_000,
    ),
    runtime_cfg=RuntimeConfig(
        end_strategy="exhaustive",
        http_client_config=HttpClientConfig(timeout=180.0, max_retries=3),
    ),
)

Providers

Provider	ModelConfig.provider	Auth env var(s)	Native structured output
OpenAI	"openai"	OPENAI_API_KEY	✅
Azure OpenAI	"azure"	AZURE_OPENAI_API_KEY, AZURE_OPENAI_ENDPOINT	✅
Anthropic	"anthropic"	ANTHROPIC_API_KEY	✅
NVIDIA NIM (openai/* models)	"nvidia"	NVIDIA_API_KEY	✅
NVIDIA NIM (other models)	"nvidia"	NVIDIA_API_KEY	❌
Google Gemini	"google-gla"	GOOGLE_API_KEY or GOOGLE_APPLICATION_CREDENTIALS	❌
Mistral AI	"mistral"	MISTRAL_API_KEY	❌

Providers without native structured output use pydantic-ai's prompt-based extraction fallback. parallel_tool_calls is automatically disabled for these providers to prevent malformed tool-call responses.

# NVIDIA NIM — openai-compatible model (native output)
ModelConfig(provider="nvidia", model="openai/gpt-oss-120b")

# NVIDIA NIM — non-OpenAI model (no native output)
ModelConfig(provider="nvidia", model="meta/llama-3.1-8b-instruct")

ReactAgent API

class ReactAgent:
    def __init__(
        self,
        config: ReactAgentConfig,
        deps_type: type[Any] | None = None,  # dependency injection type
        tools: list[Any] | None = None,       # tool functions
        toolsets: list[Any] | None = None,    # MCP server toolsets
        result_type: type[Any] = str,         # default output type
        observer: ContextObserver | None = None,
        event_loop: asyncio.AbstractEventLoop | None = None,
    ) -> None: ...

    # Execution
    async def run(self, user_prompt: UserPrompt, deps=None, output_type=None) -> Any: ...
    def run_sync(self, user_prompt: UserPrompt, deps=None, output_type=None) -> Any: ...

    # Context
    @property
    def context(self) -> ContextManager: ...
    def subscribe_context(self, observer: ContextObserver) -> None: ...
    def checkpoint(self, checkpoint_id: str | None = None) -> ContextSnapshot: ...
    def rewind(self, checkpoint_id: str) -> None: ...

    # Dynamic prompts and tools (decorator API)
    def system_prompt(self, func: Any) -> Any: ...  # wraps @agent.system_prompt(dynamic=True)
    def tool(self, func: Any) -> Any: ...            # wraps @agent.tool()

    # Advanced
    @property
    def pydantic_agent(self) -> Agent[Any, Any]: ...  # access underlying pydantic-ai Agent

output_type in run() overrides the construction-time result_type for that call only. Both are wrapped with get_output_type() to apply the provider-aware NativeOutput strategy.

Multimodal Prompts

UserPrompt = str | list[str | BinaryContent] is the accepted type for run() and run_sync(). Pass a mix of text strings and BinaryContent objects:

from pydantic_ai import BinaryContent
from akgentic.llm import ReactAgent, ReactAgentConfig, ModelConfig

agent = ReactAgent(config=ReactAgentConfig(
    model_cfg=ModelConfig(provider="openai", model="gpt-4o")
))

with open("diagram.png", "rb") as f:
    image_bytes = f.read()

result = agent.run_sync([
    "Describe what is shown in this architecture diagram.",
    BinaryContent(data=image_bytes, media_type="image/png"),
])

UserPrompt is exported from akgentic.llm so consuming layers (akgentic-agent) can annotate their own signatures without importing pydantic_ai directly.

Note: Provider support for BinaryContent varies — passing an image to a non-vision model raises a provider-level error. Multimodal turns are not JSON-serializable and are treated as ephemeral (not persisted in history replay).

Context Management

ReactAgent maintains a persistent ContextManager across calls. Message history is passed as message_history on every Agent.iter() invocation, giving the LLM full conversation continuity without manual history threading.

agent = ReactAgent(config=config)

# First turn
agent.run_sync("Start the analysis.")

# Second turn — model sees the previous exchange
agent.run_sync("Now summarise your findings.")

# Checkpoint before a risky operation
snap = agent.checkpoint("before-migration")

try:
    agent.run_sync("Apply the database migration plan.")
except Exception:
    agent.rewind("before-migration")   # restore to known-good state

ContextManager

from akgentic.llm import ContextManager

# With optional sliding window (system messages always preserved)
ctx = ContextManager(max_messages=20)

ctx.add_message(msg)
ctx.checkpoint("id", metadata={"note": "pre-flight"})
ctx.rewind("id")
ctx.get_checkpoint("id")     # → ContextSnapshot | None
ctx.list_checkpoints()       # → list[str] in creation order
ctx.subscribe(observer)
ctx.unsubscribe(observer)
ctx.clear()

Observer Pattern

from akgentic.llm import (
    ContextObserver, LlmMessageEvent, LlmCheckpointCreatedEvent,
    LlmUsageEvent, LlmSystemPromptEvent, ToolCallEvent, ToolReturnEvent,
)

class MyObserver:
    def notify_event(self, event: object) -> None:
        if isinstance(event, ToolCallEvent):
            print(f"Tool called: {event.tool_name} ({event.tool_call_id})")
        elif isinstance(event, ToolReturnEvent):
            status = "success" if event.success else "error"
            print(f"Tool returned: {event.tool_name} ({status})")
        elif isinstance(event, LlmUsageEvent):
            print(f"Usage: {event.model_name} — {event.input_tokens}in/{event.output_tokens}out")
        elif isinstance(event, LlmSystemPromptEvent):
            print(f"System prompt for run {event.run_id} ({event.content_hash[:8]}):")
            for part in event.parts:
                print(f"  [{part.dynamic_ref or 'static'}] {part.content}")
        elif isinstance(event, LlmMessageEvent):
            print(f"New message: {event.message}")
        elif isinstance(event, LlmCheckpointCreatedEvent):
            print(f"Checkpoint created: {event.snapshot.checkpoint_id}")

agent = ReactAgent(config=config, observer=MyObserver())
# or: agent.subscribe_context(MyObserver())

Events: LlmMessageEvent, LlmUsageEvent, LlmSystemPromptEvent, LlmCheckpointCreatedEvent, LlmCheckpointRestoredEvent, ToolCallEvent, ToolReturnEvent. Observers are notified synchronously — exceptions propagate to the caller.

Tool Event Observability

ToolCallEvent and ToolReturnEvent are emitted by ContextManager.add_message() after LlmMessageEvent, derived from the same message. They provide a clean observability interface for tool activity without requiring consumers to parse pydantic-ai message internals.

Part-kind → event mapping:

part_kind in message	Event emitted	Condition
tool-call	ToolCallEvent	One event per part (parallel calls → N events)
tool-return	ToolReturnEvent(success=True)	Always
retry-prompt	ToolReturnEvent(success=False)	Only when tool raised an error

Field semantics:

• tool_name — identifies which tool was called; primary routing key in observer handlers
• tool_call_id — provider-assigned identifier; correlates a ToolCallEvent with its corresponding ToolReturnEvent within the same message stream
• arguments — raw JSON string from the provider. Use json.loads(event.arguments) for structured access. Stored as str to avoid coupling to tool-specific parameter schemas.
• success — True for clean returns; False when the tool raised an error (pydantic-ai emits a retry-prompt part in that case). The return content is not carried in ToolReturnEvent; it is already in the accompanying LlmMessageEvent.

Emission ordering: LlmMessageEvent always fires first. Tool events follow immediately. A consumer receiving ToolCallEvent can safely assume the full message is already in context.

System Prompt Rendering Events

pydantic-ai re-evaluates dynamic system prompts (date, roster, role profiles, mailbox notices, …) in place before each model call, so the rendering actually sent to the model can change on runs 2+ without any LlmMessageEvent being emitted. LlmSystemPromptEvent records that effective rendering so observers (traces, frontends) can show exactly what the model saw on each run.

LlmSystemPromptEvent payload:

Field	Type	Description
run_id	str	The ReactAgent run ID this rendering belongs to — correlates with the run's LlmMessageEvent / ToolCallEvent / LlmUsageEvent
parts	tuple[SystemPromptPartSnapshot, ...]	Full rendering in model order — self-contained, not a diff
content_hash	str	sha256 hex over the ordered (dynamic_ref, content) pairs; carried in the event so dedup state can be re-seeded on restore without re-hashing

SystemPromptPartSnapshot fields:

Field	Type	Description
dynamic_ref	str | None	Function name for dynamic parts (registered via @agent.system_prompt(dynamic=True)); None for static parts
content	str	Rendered text actually sent to the model for this part

Emission semantics: emitted by ContextManager.record_system_prompt(run_id), which ReactAgent calls once per completed run after pydantic-ai's in-place re-evaluation has produced the rendering. It scans the first ModelRequest's system parts, hashes the ordered (dynamic_ref, content) pairs, and emits only when the content hash changed since the previous run — run 1 emits via the None → hash transition; an unchanged rendering on later runs emits nothing, so the log does not grow with every run; a context with no system parts emits nothing. The event store stays strictly append-only: emission only appends, and restoring an agent re-seeds the dedup hash (via seed_system_prompt_hash) without re-emitting an unchanged rendering.

Usage — label each block by its source and render the text the model saw:

from akgentic.llm import LlmSystemPromptEvent

class SystemPromptTracer:
    def notify_event(self, event: object) -> None:
        if not isinstance(event, LlmSystemPromptEvent):
            return
        print(f"System prompt @ run {event.run_id} (hash {event.content_hash[:8]})")
        for snapshot in event.parts:
            label = snapshot.dynamic_ref or "static"
            print(f"  [{label}] {snapshot.content}")

Context Compaction

Long-running agents accumulate conversation history that eventually approaches the model's context window. Compaction folds the conversation into a summary, preserving only the system prompt — the default summarize strategy replaces the entire non-system history with one summary (no verbatim tail); clear drops the history outright so the system prompt regenerates on the next run. Both are event-sourced: the ContextManager emits a single primitive event describing what changed (counts + summary text), never the replaced ModelMessage objects — so the log round-trips through the generic serializer and any subscriber can fold the same change client-side.

Compaction can fire automatically (usage-based: when the provider-reported input tokens cross trigger_ratio × context_length, no tokenizer required) or on demand via the agent's compact / clear commands.

Compaction & Clear Events

LlmContextCompactedEvent — emitted when history is folded into a summary:

Field	Type	Description
run_id	str | None	ReactAgent run the compaction belongs to; None if outside a run
strategy_id	str	Resolved strategy id (registry id or FQCN) that produced the summary
summary	str	Summary text that replaced the folded messages
replaced_message_count	int	Non-system messages folded — observability only; the summarize fold drops all non-system content regardless of this count
summarizer_prompt_version	str	Version id selecting the summarizer instructions — see Overriding the Summarizer Prompt
tokens_before	int | None	Input-token estimate before compaction; None if unknown
tokens_after	int | None	Post-compaction context-size estimate; None if the strategy doesn't report one

LlmContextClearedEvent — emitted when history is dropped without summarizing:

Field	Type	Description
run_id	str | None	ReactAgent run the clear belongs to; None if outside a run
cleared_message_count	int	Number of messages dropped from context

Both are append-only: a subscriber reconstructs the resulting context by folding the event over its own message log. For summarize the fold is full + part-level — keep only the system-prompt parts (the first request is rebuilt system-parts-only, so a user prompt fused into it by pydantic-ai is folded away too) and insert the single summary; sliding_window keeps the last keep_recent_messages; clear resets to empty. The fold no longer depends on replaced_message_count (it is observability-only).

Usage — observe compaction/clear alongside the other LLM events:

from akgentic.llm import LlmContextCompactedEvent, LlmContextClearedEvent

class CompactionTracer:
    def notify_event(self, event: object) -> None:
        if isinstance(event, LlmContextCompactedEvent):
            print(
                f"compacted @ run {event.run_id}: folded {event.replaced_message_count} msg(s) "
                f"via '{event.strategy_id}' ({event.tokens_before} → {event.tokens_after} tok est.)"
            )
            print(f"  summary: {event.summary[:120]}…")
        elif isinstance(event, LlmContextClearedEvent):
            print(f"cleared @ run {event.run_id}: dropped {event.cleared_message_count} msg(s)")

agent = ReactAgent(config=config, observer=CompactionTracer())
# or: agent.subscribe_context(CompactionTracer())

Compaction Strategies

The strategy is selected by CompactionConfig.strategy — a registry id or a dotted FQCN. Built-ins:

strategy	Behaviour	Calls an LLM?
"summarize" (default)	Replaces the entire non-system history with one summary via an awaited LLM call (system prompts kept, part-level — a user prompt fused into the first system request is folded away); no verbatim tail. Degrades to a truncation marker if the summarizer errors. Ignores keep_recent_messages.	Yes
"sliding_window"	Deterministic head-drop: keeps the last keep_recent_messages verbatim and folds the rest behind a marker, no LLM.	No
"none"	No-op: never folds a message.	No

Configure via CompactionConfig (nested in ReactAgentConfig):

from akgentic.llm import CompactionConfig

cfg = CompactionConfig(
    strategy="summarize",        # or "sliding_window", "none", or "my.module.MyStrategy"
    auto_trigger=True,           # usage-based auto-compaction
    trigger_ratio=0.85,          # fire when input tokens ≥ 0.85 × context_length
    keep_recent_messages=4,      # trailing messages kept verbatim — sliding_window only (summarize ignores it)
    summary_target_tokens=2000,  # token budget the summarizer aims for
    summarizer_prompt_version="v1",
)

Custom strategies (open extension). A CompactionStrategy is any object with async def compact(self, messages) -> CompactionResult. Register a factory in the public, mutable COMPACTION_STRATEGIES registry before building an agent, or reference a class by its dotted FQCN — the resolver imports it via stdlib importlib (akgentic-llm imports no sibling package):

from akgentic.llm import COMPACTION_STRATEGIES, CompactionConfig, CompactionResult

class KeepLastOnly:
    async def compact(self, messages):
        return CompactionResult(summary="", replaced_message_count=max(0, len(messages) - 1))

# (a) register a factory under a short id...
COMPACTION_STRATEGIES["keep_last"] = lambda cfg, model_cfg, http_client: KeepLastOnly()
cfg = CompactionConfig(strategy="keep_last")

# (b) ...or point strategy at a dotted FQCN — no registration needed
cfg = CompactionConfig(strategy="my_package.compaction.KeepLastOnly")

Overriding the Summarizer Prompt

The summarize strategy ships a domain-agnostic default system prompt. The prompt text is not stored on CompactionConfig — that config is serialized into every agent's start event, so embedding a multi-line prompt there would duplicate it across the event log. Instead the config carries only a small summarizer_prompt_version id, and the text lives in the public, mutable SUMMARY_INSTRUCTIONS registry keyed by that id (open-extension precedent: COMPACTION_STRATEGIES). The version id is also recorded on each LlmContextCompactedEvent for traceability.

Override programmatically — using the installed package, no source fork — before any agent is built:

from akgentic.llm import SUMMARY_INSTRUCTIONS, CompactionConfig

# (a) replace the default in place — every "v1" agent picks it up
SUMMARY_INSTRUCTIONS["v1"] = "You are a summarizer for legal documents. Preserve …"

# (b) register a named variant and select it per agent (the id is what lands in the event)
SUMMARY_INSTRUCTIONS["legal"] = "You are a summarizer for legal documents. Preserve …"
cfg = CompactionConfig(strategy="summarize", summarizer_prompt_version="legal")

An unknown summarizer_prompt_version falls back to the built-in default. For deployment-driven configuration (env / .env), a server's wiring layer can seed SUMMARY_INSTRUCTIONS from its settings at startup — keeping the prompt a process-level config that never enters the per-agent event stream.

Cost Tracking and Aggregation

akgentic-llm emits an LlmUsageEvent for every ModelResponse received from a provider. These events carry per-request token counts and can be aggregated into hierarchical cost summaries using aggregate_usage().

Pricing Table

Model pricing is externalized in pricing.yaml (bundled with the package). It covers Anthropic (Claude Sonnet 4, Claude Opus 4) and OpenAI (GPT-4.1 family, GPT-4o family, GPT-5 family) with per-1M-token rates for input, output, cache_read, and cache_write. The table is loaded once at import time into the PRICING dict.

Pricing resolution uses substring matching against model names (longest key first), so versioned names like "claude-sonnet-4-20250514" match the "claude-sonnet-4-20250514" key, and "gpt-4.1-mini-2025-12-11" matches "gpt-4.1-mini" before "gpt-4.1".

Aggregation

from akgentic.llm import LlmUsageEvent, aggregate_usage

# Collect events from an observer
events: list[LlmUsageEvent] = my_observer.collected_events

# Aggregate totals and per-model breakdown
summary = aggregate_usage(events)
print(f"Total cost: ${summary.total_cost_usd:.4f}")
print(f"Input tokens: {summary.total_input_tokens}")
for model_name, usage in summary.by_model.items():
    print(f"  {model_name}: ${usage.estimated_cost_usd:.4f}")

# Include per-run breakdown
summary = aggregate_usage(events, by_run=True)
for run in summary.runs:
    print(f"Run {run.run_id}: ${run.total_cost_usd:.4f}")

Data Models

Model	Description
LlmUsageEvent	Frozen dataclass emitted per ModelResponse — carries run_id, model_name, provider_name, token counts, and requests
ModelUsage	Aggregated tokens and estimated cost for a single model
RunUsageSummary	Per-run summary with per-model breakdown
AgentUsageSummary	Top-level summary with by_model, optional runs, and grand totals

Prompts

PromptTemplate

Config-time {placeholder} rendering. Used by AgentConfig.prompt in akgentic-agent:

from akgentic.llm import PromptTemplate

tpl = PromptTemplate(
    template="You are {role}.\n\nInstructions: {instructions}",
    params={"role": "the Librarian", "instructions": "Extract structured data."},
)
print(tpl.render())
# → "You are the Librarian.\n\nInstructions: Extract structured data."

Dynamic System Prompts

Register callables that are evaluated fresh on every LLM call:

from akgentic.llm import ReactAgent, ReactAgentConfig, ModelConfig
from akgentic.llm import current_datetime_prompt, json_output_reminder_prompt

agent = ReactAgent(config=ReactAgentConfig(
    model_cfg=ModelConfig(provider="openai", model="gpt-4o")
))

# Built-in utilities
agent.system_prompt(current_datetime_prompt)       # "The current date and time is …"
agent.system_prompt(json_output_reminder_prompt)   # reminder to output JSON only

# Custom prompt
@agent.system_prompt
def workspace_context(ctx: Any) -> str:
    return f"Working directory: {get_current_workspace()}"

Development

Prerequisites

• Python 3.12+
• uv package manager

Setup

uv sync --all-packages --all-extras

Commands

# Run tests
uv run pytest packages/akgentic-llm/tests/

# Run tests with coverage
uv run pytest packages/akgentic-llm/tests/ --cov=akgentic.llm --cov-fail-under=80

# Lint
uv run ruff check packages/akgentic-llm/src/

# Format
uv run ruff format packages/akgentic-llm/src/

# Type check
uv run mypy packages/akgentic-llm/src/

CI Pipeline

Every pull request runs the full quality gate via GitHub Actions (.github/workflows/ci.yml):

Step	Command	Gate
Type check	mypy packages/akgentic-llm/src/ (strict, Python 3.12)	Zero errors
Lint	ruff check packages/akgentic-llm/src/	Zero errors
Tests	pytest packages/akgentic-llm/tests/ --cov=akgentic.llm --cov-fail-under=80	All pass, ≥ 80% coverage

The CI badge at the top of this README reflects the current state of master. PRs are blocked from merging until all steps are green.

Project Structure

src/akgentic/llm/
    __init__.py     # Public API exports
    agent.py        # ReactAgent, UsageLimitError, UserPrompt type alias
    config.py       # ModelConfig, UsageLimits, HttpClientConfig, RuntimeConfig, ReactAgentConfig
    context.py      # ContextManager, ContextSnapshot
    event.py        # LlmMessageEvent, LlmUsageEvent, LlmCheckpoint*Event,
                    #   LlmSystemPromptEvent, SystemPromptPartSnapshot,
                    #   ToolCallEvent, ToolReturnEvent, ContextObserver protocol
    pricing.py      # PRICING dict, ModelUsage, RunUsageSummary, AgentUsageSummary,
                    #   aggregate_usage()
    pricing.yaml    # Externalized per-1M-token pricing table (Anthropic + OpenAI)
    prompts.py      # PromptTemplate, current_datetime_prompt, json_output_reminder_prompt
    providers.py    # create_model(), create_http_client(), get_output_type(),
                    #   create_model_settings(), _supports_native_output()
tests/              # Tests organised by module

License

See the repository root for license information.