promptflow.tracing module#

class promptflow.tracing.ThreadPoolExecutorWithContext(max_workers=None, thread_name_prefix='', initializer=None, initargs=())#

Bases: ThreadPoolExecutor

promptflow.tracing.start_trace(*, resource_attributes: Optional[dict] = None, collection: Optional[str] = None, **kwargs)#

Promptflow instrumentation.

Parameters:

  • resource_attributes (Optional[dict]) – Specify the resource attributes for current process.

  • collection (Optional[str]) – Specify the collection for current tracing.

promptflow.tracing.trace(func: Optional[Callable] = None) → Callable#

A decorator to add trace to a function.

When a function is wrapped by this decorator, the function name, inputs, outputs, start time, end time, and error (if any) will be recorded.

It can be used for both sync and async functions. For sync functions, it will return a sync function. For async functions, it will return an async function.

Parameters:

func (Callable) – The function to be traced.

Returns:

The wrapped function with trace enabled.

Return type:

Callable

Examples:

Synchronous function usage:

@trace
def greetings(user_id):
    name = get_name(user_id)
    return f"Hello, {name}"

Asynchronous function usage:

@trace
async def greetings_async(user_id):
    name = await get_name_async(user_id)
    return f"Hello, {name}"

Subpackages#