"""Development decorators for enhancing bot functionality and debugging.
This module provides decorators and utilities for:
- Lazy function implementation using LLMs (@lazy)
- Post-mortem debugging on exceptions (@debug_on_error)
- HTTP logging filters for cleaner output
The primary features are:
- @lazy: Generates function implementations at runtime using LLM
- @debug_on_error: Launches pdb debugger on exceptions
- NoHTTPFilter: Filters out HTTP-related logging noise
Example:
from bots.dev.decorators import lazy, debug_on_error
@lazy("Implement a quicksort algorithm")
def sort_list(items: list) -> list:
pass # Implementation will be generated by LLM
@debug_on_error
def risky_operation():
# Will launch debugger if this raises an exception
process_data()
"""
import os
import sys
import ast
import inspect
import logging
import traceback
import textwrap
from functools import wraps
from typing import Any, Callable, Optional
from bots.utils.helpers import remove_code_blocks
from bots.foundation.base import Bot
from bots import AnthropicBot
logging.basicConfig(level=logging.WARNING, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
[docs]
class NoHTTPFilter(logging.Filter):
"""Filters out logging records containing 'response' in the logger name.
Use when you need to reduce noise from HTTP-related logging in the application.
This filter helps prevent logging output from being flooded with HTTP response
logs while still maintaining other important log messages.
Inherits from:
logging.Filter: Base class for logging filters
Attributes:
name (str): The name of the filter (inherited from logging.Filter)
Example:
logger = logging.getLogger(__name__)
http_filter = NoHTTPFilter()
logger.addFilter(http_filter)
"""
[docs]
def filter(self, record: logging.LogRecord) -> bool:
"""Check if the log record should be filtered.
Parameters:
record (logging.LogRecord): The log record to be checked
Returns:
bool: True if the record should be kept (doesn't contain 'response'),
False if it should be filtered out
"""
return 'response' not in record.name.lower()
logger.addFilter(NoHTTPFilter())
[docs]
def lazy(prompt: Optional[str]=None, bot: Optional[Bot]=None, context: Optional[str]=None) -> Callable:
"""Decorator that lazily implements a function using an LLM at runtime.
Use when you need to generate function implementations dynamically using an LLM.
The implementation will be generated on first call and persisted to the source file.
Parameters:
prompt (Optional[str]): Additional instructions for the LLM about how to implement
the function. Defaults to empty string.
bot (Optional[Bot]): The bot instance to use for implementation. Defaults to
a new AnthropicBot instance.
context (Optional[str]): Level of context to provide to the LLM. Options are:
- 'None': No additional context
- 'low': Only the containing class
- 'medium': The entire current file
- 'high': Current file and interfaces of other files in directory
- 'very high': All Python files in directory
Defaults to 'None'.
Returns:
Callable: A decorator function that wraps the target function
Example:
@lazy("Sort using a funny algorithm. Name variables as though you're a clown.")
def sort(arr: list[int]) -> list[int]:
pass
"""
def decorator(func: Callable) -> Callable:
"""Inner decorator function that sets up the lazy implementation.
This decorator initializes the lazy implementation environment by:
1. Setting up default bot if none provided
2. Setting up default prompt if none provided
3. Setting up default context level if none provided
Parameters:
func (Callable): The function to be lazily implemented
Returns:
Callable: A wrapped version of the function that will be implemented on first call
Note:
The actual implementation is generated in the wrapper function on first call.
"""
nonlocal bot, prompt, context
if bot is None:
bot = AnthropicBot(name='Claude')
if prompt is None:
prompt = ''
if context is None:
context = 'None'
def wrapper(*args: Any, **kwargs: Any) -> Any:
"""Wrapper function that handles lazy function implementation.
This wrapper checks if the function has been implemented. If not, it:
1. Gets appropriate context based on context_level
2. Generates implementation using the LLM
3. Updates the source file with the new implementation
4. Executes the new implementation
Parameters:
*args (Any): Positional arguments to pass to the implemented function
**kwargs (Any): Keyword arguments to pass to the implemented function
Returns:
Any: The result of the implemented function
"""
nonlocal func
if not hasattr(wrapper, 'initialized') or not wrapper.initialized:
function_name: str = func.__name__
logger.debug(f'Initializing lazy function: {function_name}')
context_content = _get_context(func, context)
instructions: str = textwrap.dedent(
"""Please fill out the following
function definition according to the following requirements.
Respond only with the code in a single code block. Include all
import statements inside the function definition. Remove the lazy
decorator. Respond only with the function definition, including
any new decorators and docstring. Include 'gen by @lazy' in the
docstring. Use PEP8 convention with type hints for all variables."""
)
complete_prompt: str = textwrap.dedent(
f"""
{instructions}
{prompt}
{context_content}
{function_name}{str(inspect.signature(func))}"""
)
response: str = bot.respond(complete_prompt)
function_code, _ = remove_code_blocks(response)
function_code = function_code[0]
logger.debug(f'Generated function code:\n{function_code}')
source_file: str = inspect.getfile(func)
logger.debug(f'Source file: {source_file}')
with open(source_file, 'r') as file:
source_lines: str = file.read()
logger.debug(f'Original source file content:\n{source_lines}')
source_tree: ast.AST = ast.parse(source_lines)
class FunctionReplacer(ast.NodeTransformer):
"""AST transformer that replaces a function definition with new code.
This transformer walks the AST and replaces a specific function's
definition with new code while preserving the rest of the module.
Attributes:
function_name (str): Name of the function to be replaced
new_code (str): New implementation code to replace the function with
"""
def __init__(self, function_name: str, new_code: str):
self.function_name = function_name
self.new_code = new_code
def visit_FunctionDef(self, node: ast.FunctionDef) -> ast.AST:
if node.name == self.function_name:
logger.debug(f'Replacing function: {self.function_name}')
new_node: ast.AST = ast.parse(self.new_code).body[0]
return new_node
return node
function_replacer = FunctionReplacer(function_name, function_code)
new_tree: ast.AST = function_replacer.visit(source_tree)
ast.fix_missing_locations(new_tree)
new_source_lines: str = ast.unparse(new_tree)
logger.debug(f'New source file content:\n{new_source_lines}')
with open(source_file, 'w') as file:
file.write(new_source_lines)
logger.debug('Updated source file written')
exec(function_code, globals())
func = globals()[function_name]
wrapper.initialized = True
logger.debug(f'Lazy function {function_name} initialized')
return func(*args, **kwargs)
wrapper.initialized = False
return wrapper
return decorator
[docs]
def _get_context(func: Callable, context_level: str) -> str:
"""Retrieves contextual code based on the specified context level.
Use when you need to gather source code context for an LLM to understand
the environment of a function.
Parameters:
func (Callable): The function to get context for
context_level (str): Level of context to retrieve:
- 'None': Returns empty string
- 'low': Returns only the containing class
- 'medium': Returns entire current file
- 'high': Returns current file and interfaces of other files
- 'very high': Returns all Python files in directory
Returns:
str: The requested context as a string
Raises:
ValueError: If context_level is not one of the valid options
"""
if context_level == 'None':
return ''
source_file = inspect.getfile(func)
source_dir = os.path.dirname(source_file)
if context_level == 'low':
source_code = inspect.getsource(inspect.getmodule(func))
tree = ast.parse(source_code)
for node in ast.walk(tree):
if isinstance(node, ast.ClassDef):
for sub_node in node.body:
if isinstance(sub_node, ast.FunctionDef) and sub_node.name == func.__name__:
return ast.unparse(node)
return ''
elif context_level == 'medium':
with open(source_file, 'r') as file:
return file.read()
elif context_level == 'high':
context = ''
with open(source_file, 'r') as file:
context += f'Current file ({os.path.basename(source_file)}):\n{file.read()}\n\n'
for filename in os.listdir(source_dir):
if filename.endswith('.py') and filename != os.path.basename(source_file):
file_path = os.path.join(source_dir, filename)
context += f'Interface of {filename}:\n{_get_py_interface(file_path)}\n\n'
return context
elif context_level == 'very high':
context = ''
for filename in os.listdir(source_dir):
if filename.endswith('.py'):
file_path = os.path.join(source_dir, filename)
with open(file_path, 'r') as file:
context += f'File: {filename}\n{file.read()}\n\n'
return context
else:
raise ValueError(f'Invalid context level: {context_level}')
[docs]
def _get_py_interface(file_path: str) -> str:
"""Extracts the public interface (classes and functions) from a Python file.
Use when you need to get a high-level overview of a Python module's structure
without including implementation details.
Parameters:
file_path (str): Path to the Python file to analyze
Returns:
str: A string containing the module's interface, including class and function
signatures with their docstrings
"""
def get_docstring(node: ast.AST) -> str:
"""Extract docstring from an AST node.
Parameters:
node (ast.AST): The AST node to extract docstring from
Returns:
str: The node's docstring or empty string if none exists
"""
return ast.get_docstring(node) or ''
def format_function(node: ast.FunctionDef) -> str:
"""Format a function definition with its signature and docstring.
Parameters:
node (ast.FunctionDef): The function definition node to format
Returns:
str: Formatted string containing the function signature and docstring
"""
return f'def {node.name}{ast.unparse(node.args)}:\n """{get_docstring(node)}"""\n'
def format_class(node: ast.ClassDef) -> str:
"""Format a class definition with its signature, docstring, and method interfaces.
Parameters:
node (ast.ClassDef): The class definition node to format
Returns:
str: Formatted string containing the class definition, including base classes,
docstring, and method signatures
"""
class_str = f'class {node.name}'
if node.bases:
class_str += f"({', '.join((ast.unparse(base) for base in node.bases))})"
class_str += f':\n """{get_docstring(node)}"""\n'
for item in node.body:
if isinstance(item, ast.FunctionDef):
class_str += f' {format_function(item)}\n'
return class_str
with open(file_path, 'r') as file:
tree = ast.parse(file.read())
interface = ''
for node in tree.body:
if isinstance(node, ast.ClassDef):
interface += format_class(node) + '\n'
elif isinstance(node, ast.FunctionDef):
interface += format_function(node) + '\n'
return interface.strip()
[docs]
def debug_on_error(func: Callable) -> Callable:
"""Decorator that launches post-mortem debugging on exception.
Use when you need to automatically start a debugging session when a function
raises an unhandled exception. This is particularly useful during development
and troubleshooting.
Parameters:
func (Callable): The function to wrap with debugging capabilities
Returns:
Callable: A wrapped version of the function that will launch pdb on error
Example:
@debug_on_error
def might_fail():
# If this raises an exception, you'll get a pdb prompt
risky_operation()
"""
@wraps(func)
def wrapper(*args: Any, **kwargs: Any) -> Any:
try:
return func(*args, **kwargs)
except Exception:
type, value, tb = sys.exc_info()
traceback.print_exception(type, value, tb)
print('\n--- Entering post-mortem debugging ---')
import pdb
pdb.post_mortem(tb)
return wrapper