Skip to content

Reference for llmcode/utils/completion.py

This module handles the completion functionality using an LLM API.

It provides functions to execute tasks with a timeout and to obtain completions from LLM APIs based on specified parameters.

Functions: run_with_timeout: Executes a function with a specified timeout. get_completion_openai: Retrieves a completion from the OpenAI API using a prompt.

Author: Francisco Javier Gañán License File: https://github.com/javierganan99/LLMCode/blob/main/LICENSE


llmcode.utils.completion.run_with_timeout 

run_with_timeout(target_function, args=(), kwargs={}, timeout=30)

Runs a target function with specified arguments and keyword arguments within a given timeout.

This function uses a separate thread to execute the target_function with the provided args and kwargs. If the function completes within the timeout period, its result is returned. If it exceeds the timeout, or if an exception occurs during execution, the function will log an informative message and return None.

Parameters:

Name Type Description Default
target_function callable

The function to execute.

 required
args tuple

Positional arguments to pass to the target_function (default is empty tuple).

 ()
kwargs dict

Keyword arguments to pass to the target_function (default is empty dictionary).

 {}
timeout int

The time in seconds to wait for the function execution before timing out (default is 30).

 30

Returns:

Type Description
any | None

The result of the target_function if it completes successfully; None if it times out or raises an exception.
Source code in llmcode/utils/completion.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def run_with_timeout(target_function, args=(), kwargs={}, timeout=30):
    """
    Runs a target function with specified arguments and keyword arguments within a given timeout.

    This function uses a separate thread to execute the target_function with the provided args and kwargs.
    If the function completes within the timeout period, its result is returned. If it exceeds the timeout,
    or if an exception occurs during execution, the function will log an informative message and return None.

    Args:
        target_function (callable): The function to execute.
        args (tuple, optional): Positional arguments to pass to the target_function (default is empty tuple).
        kwargs (dict, optional): Keyword arguments to pass to the target_function (default is empty dictionary).
        timeout (int, optional): The time in seconds to wait for the function execution before timing out (default is 30).

    Returns:
        (any | None): The result of the target_function if it completes successfully; None if it times out or raises an exception.
    """
    result = [None]  # A mutable container to store the function result
    exception = [None]  # A container to capture exceptions

    def target_wrapper():
        try:
            result[0] = target_function(*args, **kwargs)
        except Exception as e:
            exception[0] = e

    thread = threading.Thread(target=target_wrapper)
    thread.start()
    thread.join(timeout)
    if thread.is_alive():
        LOGGER.info(
            "%s⚠ The completion could not be done. %s response lasted more than %s seconds, which is the limit.",
            ANSI_CODE["yellow"],
            target_function.__name__,
            timeout,
        )
        return None
    if exception[0]:
        LOGGER.info(
            "%s⚠ The completion could not be done. %s raised the exception %s",
            ANSI_CODE["yellow"],
            target_function.__name__,
            exception[0],
        )
        return None
    return result[0]




llmcode.utils.completion.get_completion_openai 

get_completion_openai(prompt)

Generates a completion response from OpenAI's Chat API based on the provided prompt.

This function takes a user-defined prompt and sends it to the OpenAI completion model, returning the content of the model's response. Additional parameters for the completion can be passed using the completion_params object, which is filtered to exclude any private attributes.

Parameters:

Name Type Description Default
prompt str

The input text that the model will generate a response for.

 required

Returns:

Type Description
str

The content of the model's response.
Source code in llmcode/utils/completion.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def get_completion_openai(prompt):
    """
    Generates a completion response from OpenAI's Chat API based on the provided prompt.

    This function takes a user-defined prompt and sends it to the OpenAI completion model,
    returning the content of the model's response. Additional parameters for the completion
    can be passed using the `completion_params` object, which is filtered to exclude any
    private attributes.

    Args:
        prompt (str): The input text that the model will generate a response for.

    Returns:
        (str): The content of the model's response.
    """
    params = {
        key: value
        for key, value in vars(completion_params).items()
        if not key.startswith("__")
    }
    return (
        openai.ChatCompletion.create(
            messages=[{"role": "user", "content": prompt}], **params
        )
        .choices[0]
        .message.content
    )