LLM Module¶
LLM client module.
- class atloop.llm.LLMClient(config: AtloopConfig, workspace_root: str | None = None)[source]¶
Bases:
objectLLM client wrapper with cache-optimized history management.
- __init__(config: AtloopConfig, workspace_root: str | None = None)[source]¶
Initialize LLM client.
- Parameters:
config – atloop configuration
workspace_root – Optional workspace root path for project skills
- add_tool_results_to_history(actions: List[Dict[str, Any]], results: List[Dict[str, Any]])[source]¶
Add tool execution results to conversation history.
NOTE: This method is kept for backward compatibility but does nothing since we no longer use ChatHistory. Tool results are stored in Memory and included in the next memory summary.
- Parameters:
actions – List of actions that were executed
results – List of tool execution results
- build_prompt(goal: str, budget: Dict[str, int], state_summary: str | None = None, project_profile: str | None = None, relevant_files: str | None = None, recent_error: str | None = None, current_diff: str | None = None, test_results: str | None = None, verification_success: bool | None = None) str[source]¶
Build complete prompt from templates (DEPRECATED - for backward compatibility).
This method is kept for backward compatibility but should be replaced with build_user_message() for cache optimization.
- Parameters:
goal – Task goal
budget – Budget dictionary
state_summary – State summary
project_profile – Project profile
relevant_files – Relevant file snippets
recent_error – Recent error
current_diff – Current diff
test_results – Test results
verification_success – Verification success status
- Returns:
Complete prompt string
- build_user_message(goal: str, budget: Dict[str, int], state_summary: str | None = None, memory_context: str | None = None, project_profile: str | None = None, relevant_files: str | None = None, recent_error: str | None = None, current_diff: str | None = None, test_results: str | None = None, verification_success: bool | None = None) str[source]¶
Build user message from context (append-only mode for cache optimization).
This method builds a user message that will be appended to the conversation history. The system prompt is fixed and never changes, preserving LLM cache.
- Parameters:
goal – Task goal
budget – Budget dictionary
state_summary – State summary
project_profile – Project profile
relevant_files – Relevant file snippets
recent_error – Recent error
current_diff – Current diff
test_results – Test results
verification_success – Verification success status
- Returns:
User message string (to be appended to history)
- generate_tool_schema() str[source]¶
Generate comprehensive tool schema description for prompt.
Dynamically extracts detailed descriptions from tool classes, including: - Usage guidelines - Important warnings - Parameters - Examples - When to use vs other tools
- Returns:
Formatted tool schema description string
- load_prompt_template(template_name: str) str[source]¶
Load prompt template using PromptLoader.
- Parameters:
template_name – Template name (system, developer)
- Returns:
Template content
- plan_and_act(user_message: str, max_retries: int = 2, stream_callback: Callable[[str], None] | None = None, step: int | None = None, task_id: str | None = None) Tuple[ActionJSON | None, str | None, Dict[str, Any], str | None, Dict[str, str]][source]¶
Call LLM to plan and generate actions.
Memory-only mode (no ChatHistory): - Fixed system prompt (set once, never changes - preserves cache) - Each call is independent with full context in user_message - All history is managed through Memory, not conversation history - Uses lexilux 2.0.0 stream() for ensuring complete responses
- Parameters:
user_message – Required user message content containing all context (including memory summary)
max_retries – Maximum retry attempts if JSON parsing fails
stream_callback – Optional callback function to receive streaming output chunks
step – Optional step number for event emission
task_id – Optional task ID for event emission
- Returns:
Tuple of (ActionJSON or None, error_message, usage_info, full_output_text, file_contents_dict) file_contents_dict maps placeholder names to actual file content
- class atloop.llm.ActionJSON(actions: List[Dict[str, Any]], stop_reason: str, current_step_thoughts: str | None = None, plan: List[str] | None = None, result_message: str | None = None)[source]¶
Bases:
objectAction JSON data structure.
Design principle: Fail Fast - Data validation happens at construction time - Invalid data is rejected immediately with clear error messages - Downstream code can trust that ActionJSON instances are valid
- __init__(actions: List[Dict[str, Any]], stop_reason: str, current_step_thoughts: str | None = None, plan: List[str] | None = None, result_message: str | None = None)[source]¶
Initialize Action JSON.
- Parameters:
actions – List of action dictionaries
stop_reason – Stop reason (continue, done, fail)
current_step_thoughts – Optional current step thoughts (not a summary)
plan – Optional plan steps
result_message – Optional result message
- Raises:
ActionJSONValidationError – If data is invalid
- classmethod from_dict(data: Dict[str, Any], validate: bool = True, tool_registry: ToolRegistry | None = None) ActionJSON[source]¶
Create from dictionary with validation.
Design principle: Validate at the boundary - Data entering the system is validated immediately - Invalid data is rejected with clear error messages - Downstream code can trust the data structure
- Parameters:
data – Dictionary containing action JSON data
validate – Whether to validate the data (default: True)
tool_registry – Optional ToolRegistry instance for dynamic tool validation. If provided, validates tool existence and delegates argument validation to tool.validate_args(). If not provided, only performs structural checks.
- Returns:
ActionJSON instance
- Raises:
ActionJSONValidationError – If data is invalid and validate=True
TypeError – If data is not a dictionary
- exception atloop.llm.ActionJSONValidationError(message: str, data: Dict[str, Any] | None = None)[source]¶
Bases:
ValueErrorException raised when ActionJSON validation fails.
- atloop.llm.parse_action_json(text: str, max_retries: int = 2, tool_registry: ToolRegistry | None = None) Tuple[ActionJSON | None, str | None, Dict[str, str]][source]¶
Parse Action JSON from text with improved error handling.
Also extracts file contents from placeholders (TYPE_descriptive-name format) that follow the JSON in the format: —((WRITE_FILE_CONTENT_descriptive-name))— <file content> —((SHELL_COMMAND_descriptive-name))— <command> …
Tries multiple strategies: 1. Direct JSON parsing 2. Extract JSON from code blocks (
`json or `) 3. Extract JSON by matching braces (handling strings) 4. Fix common JSON errors 5. Use json-repair if available 6. Use json5 if available- Parameters:
text – Text containing JSON and optionally file contents
max_retries – Maximum number of retries (unused, kept for compatibility)
tool_registry – Optional ToolRegistry instance for dynamic tool validation. If provided, validates tool existence and delegates argument validation to tool.validate_args(). If not provided, only performs structural checks.
- Returns:
Tuple of (ActionJSON or None, error_message, file_contents_dict) file_contents_dict maps placeholder names (e.g., “WRITE_FILE_CONTENT_descriptive-name”) to actual content
- atloop.llm.validate_action_json(data: Dict[str, Any], tool_registry: ToolRegistry | None = None) Tuple[bool, str | None][source]¶
Validate Action JSON structure with detailed error messages.
This function performs structural validation only. Tool-specific argument validation is delegated to each tool’s validate_args() method.
- Parameters:
data – Action JSON dictionary
tool_registry – Optional ToolRegistry instance for dynamic tool validation. If provided, validates tool existence and delegates argument validation to tool.validate_args(). If not provided, only performs structural checks.
- Returns:
Tuple of (is_valid, error_message)
LLMClient¶
- class atloop.llm.client.LLMClient(config: AtloopConfig, workspace_root: str | None = None)[source]¶
Bases:
objectLLM client wrapper with cache-optimized history management.
- __init__(config: AtloopConfig, workspace_root: str | None = None)[source]¶
Initialize LLM client.
- Parameters:
config – atloop configuration
workspace_root – Optional workspace root path for project skills
- generate_tool_schema() str[source]¶
Generate comprehensive tool schema description for prompt.
Dynamically extracts detailed descriptions from tool classes, including: - Usage guidelines - Important warnings - Parameters - Examples - When to use vs other tools
- Returns:
Formatted tool schema description string
- load_prompt_template(template_name: str) str[source]¶
Load prompt template using PromptLoader.
- Parameters:
template_name – Template name (system, developer)
- Returns:
Template content
- build_user_message(goal: str, budget: Dict[str, int], state_summary: str | None = None, memory_context: str | None = None, project_profile: str | None = None, relevant_files: str | None = None, recent_error: str | None = None, current_diff: str | None = None, test_results: str | None = None, verification_success: bool | None = None) str[source]¶
Build user message from context (append-only mode for cache optimization).
This method builds a user message that will be appended to the conversation history. The system prompt is fixed and never changes, preserving LLM cache.
- Parameters:
goal – Task goal
budget – Budget dictionary
state_summary – State summary
project_profile – Project profile
relevant_files – Relevant file snippets
recent_error – Recent error
current_diff – Current diff
test_results – Test results
verification_success – Verification success status
- Returns:
User message string (to be appended to history)
- build_prompt(goal: str, budget: Dict[str, int], state_summary: str | None = None, project_profile: str | None = None, relevant_files: str | None = None, recent_error: str | None = None, current_diff: str | None = None, test_results: str | None = None, verification_success: bool | None = None) str[source]¶
Build complete prompt from templates (DEPRECATED - for backward compatibility).
This method is kept for backward compatibility but should be replaced with build_user_message() for cache optimization.
- Parameters:
goal – Task goal
budget – Budget dictionary
state_summary – State summary
project_profile – Project profile
relevant_files – Relevant file snippets
recent_error – Recent error
current_diff – Current diff
test_results – Test results
verification_success – Verification success status
- Returns:
Complete prompt string
- plan_and_act(user_message: str, max_retries: int = 2, stream_callback: Callable[[str], None] | None = None, step: int | None = None, task_id: str | None = None) Tuple[ActionJSON | None, str | None, Dict[str, Any], str | None, Dict[str, str]][source]¶
Call LLM to plan and generate actions.
Memory-only mode (no ChatHistory): - Fixed system prompt (set once, never changes - preserves cache) - Each call is independent with full context in user_message - All history is managed through Memory, not conversation history - Uses lexilux 2.0.0 stream() for ensuring complete responses
- Parameters:
user_message – Required user message content containing all context (including memory summary)
max_retries – Maximum retry attempts if JSON parsing fails
stream_callback – Optional callback function to receive streaming output chunks
step – Optional step number for event emission
task_id – Optional task ID for event emission
- Returns:
Tuple of (ActionJSON or None, error_message, usage_info, full_output_text, file_contents_dict) file_contents_dict maps placeholder names to actual file content
- reset_history()[source]¶
Reset conversation history (for new task).
NOTE: This method is kept for backward compatibility but does nothing since we no longer use ChatHistory. All history is managed through Memory.
- add_tool_results_to_history(actions: List[Dict[str, Any]], results: List[Dict[str, Any]])[source]¶
Add tool execution results to conversation history.
NOTE: This method is kept for backward compatibility but does nothing since we no longer use ChatHistory. Tool results are stored in Memory and included in the next memory summary.
- Parameters:
actions – List of actions that were executed
results – List of tool execution results
ActionJSON¶
- class atloop.llm.schema.ActionJSON(actions: List[Dict[str, Any]], stop_reason: str, current_step_thoughts: str | None = None, plan: List[str] | None = None, result_message: str | None = None)[source]¶
Bases:
objectAction JSON data structure.
Design principle: Fail Fast - Data validation happens at construction time - Invalid data is rejected immediately with clear error messages - Downstream code can trust that ActionJSON instances are valid
- __init__(actions: List[Dict[str, Any]], stop_reason: str, current_step_thoughts: str | None = None, plan: List[str] | None = None, result_message: str | None = None)[source]¶
Initialize Action JSON.
- Parameters:
actions – List of action dictionaries
stop_reason – Stop reason (continue, done, fail)
current_step_thoughts – Optional current step thoughts (not a summary)
plan – Optional plan steps
result_message – Optional result message
- Raises:
ActionJSONValidationError – If data is invalid
- classmethod from_dict(data: Dict[str, Any], validate: bool = True, tool_registry: ToolRegistry | None = None) ActionJSON[source]¶
Create from dictionary with validation.
Design principle: Validate at the boundary - Data entering the system is validated immediately - Invalid data is rejected with clear error messages - Downstream code can trust the data structure
- Parameters:
data – Dictionary containing action JSON data
validate – Whether to validate the data (default: True)
tool_registry – Optional ToolRegistry instance for dynamic tool validation. If provided, validates tool existence and delegates argument validation to tool.validate_args(). If not provided, only performs structural checks.
- Returns:
ActionJSON instance
- Raises:
ActionJSONValidationError – If data is invalid and validate=True
TypeError – If data is not a dictionary