agent

agent

Agent — orchestrates RAG + tool-calling loop.

Agent

Orchestrates RAG retrieval, tool calling, and LLM interaction.

Manages a single conversation session with its own history, system prompt (with configurable political axes), and knowledge base. Supports both synchronous (process) and async streaming (process_stream) interfaces.

The agent loop: 1. Retrieves relevant chunks from the knowledge base (RAG) 2. Builds the message context (system prompt + history + user query) 3. Streams the LLM response, handling tool calls iteratively 4. Returns the final response with source citations and tool metadata

__init__

__init__(
    embedder: Embedder,
    kb: KnowledgeBase,
    system_prompt: str = "",
    axes: dict | None = None,
    keys: dict[str, str | None] | None = None,
)

Initialize an Agent instance.

Parameters:

Name	Type	Description	Default
embedder	Embedder	Embedder instance for RAG retrieval.	required
kb	KnowledgeBase	KnowledgeBase instance (may be unloaded; loaded on first query).	required
system_prompt	str	Full system prompt string. If empty, uses build_system_prompt() with the given axes.	''
axes	dict | None	Political axes dict. Defaults to module-level AXES config.	None
keys	dict[str, str | None] | None	Per-user API keys dict with 'deepseek', 'tavily', 'scrapingant'.	None

set_axes

set_axes(axes: dict) -> None

Update the agent's political axes and rebuild the system prompt.

Parameters:

Name	Type	Description	Default
axes	dict	Dict with keys "ethical", "alignment", "social" and float values in [-1.0, 1.0].	required

set_http_client

set_http_client(client: AsyncClient | None) -> None

Set or clear the HTTP client for connection reuse (Phase 2+).

process_stream async

process_stream(
    question: str, axes: dict | None = None
) -> AsyncGenerator[AgentEvent, None]

Async generator yielding typed events at each agent loop step.

Event sequence: SourceEvent(s) → TokenEvent(s) → ToolCallEvent/ToolResultEvent pairs → TokenEvent(s) → DoneEvent. Accepts optional axes dict to override agent's current axes for this turn.

Wraps the entire body in try/except per T-01-01 (threat model) to prevent leaking internal state via unhandled exceptions.

process

process(
    question: str, axes: dict | None = None
) -> AgentResult

Synchronous wrapper — delegates to process_stream() via asyncio.run().

Collects all streamed events and reconstructs the AgentResult contract for CLI backward compatibility. run.py calls this method unchanged.

AgentManager

Singleton manager for shared resources and per-user Agent instances.

Provides a single point of access for: - Shared Embedder and KnowledgeBase instances (memory-efficient) - Shared httpx.AsyncClient for connection reuse - Per-user Agent instances via an AgentRegistry with LRU eviction

Thread-safe singleton pattern with double-checked locking.

__init__

__init__() -> None

Initialize the shared embedder and knowledge base.

Creates an Embedder with EMBED_DTYPE and MAX_VOCAB_SIZE from config, and a KnowledgeBase with optional mmap support. The HTTP client and agent registry are lazily initialized.

get_instance classmethod

get_instance() -> 'AgentManager'

Get or create the singleton AgentManager instance.

Thread-safe via double-checked locking.

Returns:

Type	Description
'AgentManager'	The singleton AgentManager instance.

get_client async

get_client() -> AsyncClient

Get or create the shared httpx.AsyncClient for connection reuse.

Initialized with a 120-second timeout (30-second connect timeout).

Returns:

Type	Description
AsyncClient	Shared httpx.AsyncClient instance.

get_or_create async

get_or_create(
    user_id: str,
    axes: dict | None = None,
    keys: dict[str, str | None] | None = None,
) -> Agent

Get or create an Agent instance for a specific user.

Uses the AgentRegistry for LRU-based caching with TTL-based eviction. New agents are created with the shared embedder, knowledge base, and HTTP client.

Parameters:

Name	Type	Description	Default
user_id	str	Unique user identifier.	required
axes	dict | None	Optional axis overrides for the agent.	None
keys	dict[str, str | None] | None	Optional per-user API keys dict.	None

Returns:

Type	Description
Agent	Agent instance associated with the user.

shutdown async

shutdown() -> None

Clean up shared resources.

Closes the shared httpx.AsyncClient and clears the agent registry. Should be called during application shutdown (lifespan handler).