Skip to content

appl

appl - A Prompt Programming Language.

get_parser 

get_parser(
    env_prefix: str = "",
    default_env: bool = True,
    **kwargs: Any
) -> ArgumentParser

Get an argument parser with configurable APPL configs.

Source code in src/appl/__init__.py 
93
94
95
96
97
98
99
def get_parser(
    env_prefix: str = "", default_env: bool = True, **kwargs: Any
) -> ArgumentParser:
    """Get an argument parser with configurable APPL configs."""
    parser = ArgumentParser(env_prefix=env_prefix, default_env=default_env, **kwargs)
    parser.add_argument("--appl", type=APPLConfigs, default=global_vars.configs)
    return parser

init 

init(**kwargs: Any) -> None

Overwrite APPL configs.

see default configs for more details.

Examples:

import appl

appl.init(servers={"default": "gpt-4o"})
Source code in src/appl/__init__.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
def init(**kwargs: Any) -> None:
    """Overwrite APPL configs.

    see [default configs](../setup/#default-configs) for more details.

    Examples:
        ```python
        import appl

        appl.init(servers={"default": "gpt-4o"})
        ```
    """
    logger.warning(
        f"appl.init() is not mandatory to initialize APPL for appl>=0.2.0, "
        "please remove it if you are not changing the configs."
    )
    if kwargs:
        logger.info(f"Updating APPL configs with: \n{yaml.dump(kwargs)}")
        new_configs = merge_configs(global_vars.configs, **kwargs)
        update_appl_configs(new_configs)

init_within_thread 

init_within_thread(
    log_file_prefix: Optional[str] = None,
    gen_name_prefix: Optional[str] = None,
) -> Any

Initialize APPL to work with multi-threading, including logging and tracing.

Parameters:

  • log_file_prefix (Optional[str], default: None ) –

    The prefix for the log file. Defaults to use the path of the main log file.

  • gen_name_prefix (Optional[str], default: None ) –

    The prefix for the generation name. Defaults to use the thread name.

Examples:

def function_run_in_thread():
    with appl.init_within_thread():
        # do something within the thread
Source code in src/appl/__init__.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
@contextmanager
def init_within_thread(
    log_file_prefix: Optional[str] = None, gen_name_prefix: Optional[str] = None
) -> Any:
    """Initialize APPL to work with multi-threading, including logging and tracing.

    Args:
        log_file_prefix: The prefix for the log file. Defaults to use the path of the main log file.
        gen_name_prefix: The prefix for the generation name. Defaults to use the thread name.

    Examples:
        ```python
        def function_run_in_thread():
            with appl.init_within_thread():
                # do something within the thread
        ```
    """
    handler_id = None

    try:
        thread_name = threading.current_thread().name
        logging_settings = global_vars.configs.settings.logging

        def filter_thread_record(record: Dict) -> bool:
            assert hasattr(record["thread"], "name")
            # Use prefix match to filter the log records in different threads
            name = record["thread"].name
            return name == thread_name or name.startswith(thread_name + "_")

        if logging_settings.enable_file:
            if log_file_prefix is None:
                assert (
                    global_vars.metadata.log_file is not None
                ), "should have log file set"
                thread_log_path = os.path.join(
                    global_vars.metadata.log_file[: -len(".log")] + "_logs",
                    f"{thread_name}.log",
                )
            else:
                thread_log_path = f"{log_file_prefix}_{thread_name}.log"

            log_level = (
                logging_settings.log_file.log_level or logging_settings.log_level
            )
            # The logger append to the file by default, not overwrite.
            handler_id = logger.add(
                thread_log_path,
                level=log_level,
                format=logging_settings.format,
                filter=filter_thread_record,  # type: ignore
            )
        if gen_name_prefix:
            set_gen_name_prefix(gen_name_prefix)
            # ? shall we reset the prefix after exiting the context?
            logger.info(
                f"Thread {thread_name}, set generation name prefix as: {gen_name_prefix}"
            )

        if handler_id is None:
            logger.warning("logging is not enabled")
        yield thread_log_path
    except Exception as e:
        logger.error(f"Error in thread: {e}")
        raise e
    finally:
        if handler_id:
            logger.remove(handler_id)