core.agent

Agent system modules: streaming generation, tool calling, prompt building, memory context.

Agent Classes

core.agent.agent_classes

Agent

Class of an agent that will be speaking.

The class is responsible for generating a text response and summarizing the chat history when appropriate.

fetch_url(url: str, text_length: int = 8000) -> str

Fetch a URL with curl_cffi TLS impersonation and return clean text.

Uses Chrome TLS fingerprint impersonation to bypass bot detection. Includes SSRF protection against private/internal URLs.

Parameters

url : str The full URL to fetch (http:// or https://). text_length : int, optional Maximum characters of page content to return (default 8000).

generate_text_stream(messages: list, system_prompt: str, user_input: str, tools: dict = None, **kwargs) -> tuple[list, AsyncGenerator[tuple[str, str], None]] async

Yield tuples (channel, sentence) where channel ∈ {'think', 'answer'}. This version streams tokens to the UI using the streaming protocol.

web_search(query, messages=None, reranker=None)

DuckDuckGo search returning titles and URLs only.

No page fetching — use fetch_url to read specific pages. Kept reranker param for backward compatibility (ignored).

Agent Factory

core.agent.agent_factory

Agent factory — instantiates Agent objects at runtime.

Keeps config.py free of heavy imports (chromadb, agent_classes) so that importing the config module is fast and side-effect free.

build_agents(cfg: Any, *, model_backend: Any = None, tool_host: Any = None) -> list[Any]

Build the list of Agent instances from the config module.

Parameters:

Name	Type	Description	Default
cfg	Any	Module exposing all model names, agent_config, system prompts, etc.	required
model_backend	Any	Optional ModelBackend instance passed to each Agent.	None
tool_host	Any	Required ToolHost instance for the 8-step security pipeline. Passing None raises ValueError.	None

Returns:

Type	Description
list[Any]	A list of Agent instances ready to have db_client,
list[Any]	reranker, and tts assigned by the caller.

Raises:

Type	Description
ValueError	If tool_host is None — ToolHost is mandatory for all agent creation; all tool calls must pass through the security pipeline.

Generation Service

core.agent.generation_service

Generation service — Prompt building, generation options, sentence helpers.

Extracted from main.py (Item #16: Refactor main.py into lifecycle services). Covers: calc_sentence_length, format_timestamp, has_ui_sampling_override, is_gemma4_model, selected_generation_model, build_generation_options, log_effective_generation_options, normalize_agent_think, build_contextual_information, QueueResponseConfig, select_prompt_config, t3_to, clean_math_text.

These functions are pure or depend only on the config module and agent objects. They are re-exported from main.py for backward compatibility.

build_contextual_information(agent: Any, sentence_length: int, mode_kwargs: dict) -> str

Construct the base contextual instruction block.

Minimal prompt architecture for modern models: - Personality (agent system_prompt1) first - Dialogue quality harness (shared TTS/plain-text rules) - Sentence count constraint + plaintext math - Tool use guidance (no announcement, consecutive calls ok) - Mode status appended when any modes active

build_generation_options(agent: Any, analysis_mode: bool, config: Any = None, auto_chat_mode_var: bool = False) -> dict

Build effective Ollama runtime options while preserving explicit UI overrides.

calc_sentence_length(user_voice_output: str) -> int

Determine the desired sentence length (capped at 4).

clean_math_text(text: str) -> str

Replace mathematical notation with plaintext for TTS.

format_timestamp() -> tuple[str, str]

Return weekday name and formatted datetime string.

has_ui_sampling_override(option_name: str, config: Any = None) -> bool

Return whether the UI has explicitly overridden a sampling option.

is_gemma4_model(model_name: str) -> bool

Return whether the active Ollama model is from the Gemma 4 family.

log_effective_generation_options(agent: Any, options: dict, analysis_mode: bool) -> None

Log the effective generation options for debugging.

normalize_agent_think(value, model_name: str = '')

Normalize UI/model think values into a safe routing flag.

select_prompt_config(user_voice_output: str, agent: Any, analysis_mode: bool, other_active_agents: list, config: Any = None, auto_chat_mode_var: bool = False) -> QueueResponseConfig

Determine prompt, generation options, and slicing length based on current conditions.

Mirrors the original branching logic from main.py.

selected_generation_model(agent: Any, analysis_mode: bool, think_level) -> str

Mirror Agent.generate_text_stream model routing for option defaults/logging.

t3_to(model: Any, dtype)

Cast ChatterboxTTS T3 weights to the given dtype (e.g., bfloat16).

Prompt Building

core.agent.prompt_builder

PromptBuilder — Prompt construction, system prompt management.

Extracted from core/agent_classes.py (Item #36: Agent split). Pure functions that operate on agent attributes to build prompts and context. No async I/O — these are synchronous string manipulation helpers.

adjust_system_prompt_text(current_override: str, new_instruction: str, mode: str = 'append') -> str

Adjust the system prompt override text.

Parameters

current_override : The existing system prompt override string. new_instruction : The new instruction to add or replace with. mode : "append" (add to end), "prepend" (add to beginning), or "replace".

Returns

str Updated system prompt override.

build_contextual_block(agent: Any, sentence_length: int, mode_kwargs: dict) -> str

Build the contextual instruction block for the prompt.

Delegates to core.generation_service.build_contextual_information for the actual construction. This wrapper adapts the Agent interface.

Parameters

agent : Agent instance with system_prompt1. sentence_length : Target sentence count. mode_kwargs : Dict of active modes (e.g., {"analysis_mode": "True"}).

Returns

str Full contextual instruction block.

build_mode_status_block(analysis_mode: bool = False, think_mode: str = '', mute_mode: bool = False, auto_chat_mode: bool = False, cloud_mode: bool = False, training_mode: bool = False) -> dict

Build the mode status dictionary for context injection.

Returns

dict Mapping of mode name to string representation.

build_system_prompt(agent: Any, system_prompt_override: str = '') -> str

Build the primary system prompt for an agent.

Parameters

agent : Agent instance with system_prompt1 attribute. system_prompt_override : Optional override text to append to the primary prompt.

Returns

str The primary prompt string.

format_response_format(sentence_length: int) -> str

Build the [RESPONSE FORMAT] section.

Parameters

sentence_length : Number of sentences the agent should produce.

Returns

str Response format instruction text.

format_tool_calling_instructions() -> str

Build tool calling instruction text.

Returns

str Tool calling instruction text.

Tool Invocation

core.agent.tool_invocation_service

ToolInvocationService — Tool call parsing, normalization, and execution routing.

Extracted from core/agent_classes.py (Item #36: Agent split). Provides pure functions for tool call normalization, validation, and execution result handling.

ToolExecutionContext

Context for a tool execution cycle.

Tracks the current tool call state and accumulated results.

can_execute(tool_name: str) -> tuple[bool, str | None]

Check if a tool execution is allowed.

Parameters

tool_name : Lowercase tool name to check (kept for API compatibility).

Returns

tuple[bool, str | None] Always (True, None) — no per-turn invocation limits.

record_call(tool_name: str, result: str) -> None

Record a tool call execution.

Parameters

tool_name : Name of the executed tool. result : The tool output string.

summary() -> dict

Return a summary of the execution context.

Returns

dict Summary with calls_made, web_search_count, tools_used.

build_assistant_tool_message(assistant_content: str, tool_calls: list) -> dict

Build an assistant message with tool calls for the conversation history.

Parameters

assistant_content : The assistant's text content (may be empty). tool_calls : List of normalized tool call dicts. Each may optionally carry an "id" key (preserved from Ollama) that is used as the tool-call correlation ID. If absent, a synthetic call_{i} ID is generated.

Returns

dict Message dict for appending to conversation history.

build_tool_message(role: str, tool_call_id: str, content: str) -> dict

Build a tool result message for the conversation history.

Parameters

role : Message role ("tool"). tool_call_id : The tool call correlation ID. content : The tool output string.

Returns

dict Message dict for appending to conversation history.

extract_tool_info(tool: dict) -> tuple[str, dict]

Extract tool name and arguments from a normalized tool call.

Parameters

tool : Normalized tool call dict (from normalize_tool_call).

Returns

tuple[str, dict] (tool_name_lower, arguments_dict)

format_tool_error(tool_name: str, error: Exception) -> str

Format a tool execution error for the agent.

Parameters

tool_name : Name of the failed tool. error : Exception that occurred during execution.

Returns

str Error message for the agent context.

format_tool_result(tool_name: str, result: Any) -> str

Format a tool execution result for the agent.

Parameters

tool_name : Name of the executed tool. result : Tool output (string or dict).

Returns

str Formatted result string for the agent context.

normalize_tool_call(tool_call: Any) -> dict

Convert Ollama tool call objects into plain JSON-serializable dicts.

Handles multiple input formats: - Plain dict (already normalized) - Pydantic model (has model_dump or dict method) - Attribute-based object (has .function.name, .function.arguments)

Parameters

tool_call : Raw tool call from Ollama stream.

Returns

dict Normalized tool call with "type", "function.name", "function.arguments".

should_handle_async(tool_name: str) -> bool

Determine if a tool requires async handling (not executor-offloaded).

Parameters

tool_name : Lowercase tool name.

Returns

bool True if the tool has dedicated async handling.

should_offload_to_executor(tool_name: str) -> bool

Determine if a tool should be offloaded to a thread pool executor.

Parameters

tool_name : Lowercase tool name.

Returns

bool True if the tool is blocking and should run in an executor.

validate_tool_in_schema(tool_name: str, tool_calls_active: list) -> bool

Check if a tool exists in the active schemas.

Parameters

tool_name : Lowercase tool name. tool_calls_active : List of tool schema dicts from tools/schemas.py.

Returns

bool True if the tool is registered in the active schemas.

Memory Context

core.agent.memory_context_service

MemoryContextService — Vector DB queries, embedding context, memory retrieval.

Extracted from core/agent_classes.py (Item #36: Agent split). Provides helpers for memory/vector operations used by the Agent: embedding queries, document reranking, folder-scoped retrieval.

build_query_documents(messages: list[dict], context_messages: int = 5) -> list[tuple[str, str]]

Build query-document pairs from message history for vector search.

Parameters

messages : Conversation message list (role/content dicts). context_messages : Number of recent messages to use as context.

Returns

list[tuple[str, str]] List of (query_text, document_text) pairs.

create_temp_collection_name(conversation_id: str | None = None) -> str

Generate a unique collection name for vector operations.

Parameters

conversation_id : Optional conversation ID to include in the name.

Returns

str Unique collection name string.

filter_documents_by_relevance(documents: list[tuple[str, float]], min_score: float = 0.9, max_results: int = 5) -> str

Filter and format documents by relevance score.

Parameters

documents : List of (text, score) tuples. min_score : Minimum relevance score to include. max_results : Maximum number of documents to return.

Returns

str Concatenated relevant document texts.

format_folder_context(folder_id: str, documents: str, vector_text: str) -> str

Format folder-scoped RAG context for the prompt.

Parameters

folder_id : The folder ID (for logging/debugging). documents : Retrieved document text (reranked or top-N). vector_text : Base vector text (screenshot + audio transcript, sanitized).

Returns

str Formatted folder memory context with base vector text.

format_vector_context(documents: list[str], screenshot_description: str = '', audio_transcript: str = '') -> str

Format retrieved documents into a context string for the agent.

Parameters

documents : List of document texts (from vector search or reranking). screenshot_description : Optional screenshot description text. audio_transcript : Optional audio transcript text.

Returns

str Formatted context string ready for prompt injection.

query_vector_store(db_client: Any, query_text: str, collection_name: str, n_results: int = 20) -> dict | None async

Query a ChromaDB collection for similar documents.

Offloads blocking embedding and query operations to an executor thread so the event loop stays responsive.

Parameters

db_client : ChromaDB client instance. query_text : The search query string. collection_name : ChromaDB collection name. n_results : Number of results to retrieve.

Returns

dict | None Query result dict with "documents", "distances", etc., or None on error.

rerank_documents(reranker: Any, documents: list[str], query: str, max_results: int = 3) -> str | None async

Rerank documents using a CrossEncoder model.

Offloads blocking GPU work to an executor thread.

Parameters

reranker : Reranker instance (may be None). documents : List of document texts to rerank. query : The search query text. max_results : Maximum number of results to return.

Returns

str | None Reranked document text, or None on failure.