Skip to content

log_middleware ¤

Logger Middleware

Classes:

Name Description
LoggingMiddleware

Creates a Simple logging middleware for a given task.

LoggingMiddleware ¤

Bases: TaskMiddleware

Creates a Simple logging middleware for a given task.

Example 
@openai.task(middlewares=[LoggingMiddleware])
def generate_a_poem(title: str):
    '''
    Generate a poem based on the given title
    :return: The generated poem
    '''
    return declarai.magic("poem", title)

Methods:

Name Description
__call__

Once the middleware is called, it executes the task and returns the result.
_stream

Re-streams the streaming response while adding the after sideeffects execution to the generator
after

After execution of the task, log the task details.
before

Before execution of the task, set the start time.

__call__ ¤

__call__() -> Any

Once the middleware is called, it executes the task and returns the result. Before it executes the task, it calls the before method, and after it executes the task, it calls the after method. Returns: The result of the task

Source code in src/declarai/middleware/base.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def __call__(self) -> Any:
    """
    Once the middleware is called, it executes the task and returns the result.
    Before it executes the task, it calls the `before` method, and after it executes the task, it calls the `after` method.
    Returns:
        The result of the task
    """
    self.before(self._task)
    # # If the task is streaming, handle it differently
    if self._task.operator.streaming:
        # Yield chunks from the task, then call the after method
        return self._stream()
    else:
        # Non-streaming tasks can be handled as before
        res = self._task._exec(self._kwargs)
        self.after(self._task)
        return res

_stream ¤

_stream() -> Iterator

Re-streams the streaming response while adding the after sideeffects execution to the generator Returns:

Source code in src/declarai/middleware/base.py 
27
28
29
30
31
32
33
34
35
def _stream(self) -> Iterator:
    """
    Re-streams the streaming response while adding the after sideeffects execution to the generator
    Returns:

    """
    for chunk in self._task._exec(self._kwargs):
        yield chunk
    self.after(self._task)

after ¤

after(task: TaskType)

After execution of the task, log the task details. Args: task: the task to be logged

Returns:

Type Description
Dict[str, Any]

the task details like execution time, task name, template, compiled template, result, time.
Source code in src/declarai/middleware/internal/log_middleware.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def after(self, task: TaskType):
    """
    After execution of the task, log the task details.
    Args:
        task: the task to be logged

    Returns:
        (Dict[str, Any]): the task details like execution time, task name, template, compiled template, result, time.

    """
    end_time = time() - self.start_time
    log_record = {
        "task_name": task.__name__,
        "llm_model": task.llm_response.model,
        "template": str(task.compile()),
        "call_kwargs": str(self._kwargs),
        "compiled_template": str(task.compile(**self._kwargs)),
        "result": task.llm_response.response,
        "time": end_time,
    }
    logger.info(log_record)
    print(log_record)

before ¤

before(_)

Before execution of the task, set the start time.

Source code in src/declarai/middleware/internal/log_middleware.py 
31
32
33
34
35
def before(self, _):
    """
    Before execution of the task, set the start time.
    """
    self.start_time = time()