Skip to main content

weave.trace.weave_client

API Overview

Classes​

class WeaveClient​

method __init__​

__init__(
    entity: str,
    project: str,
    server: TraceServerInterface,
    ensure_project_exists: bool = True
)

method add_cost​

add_cost(
    llm_id: str,
    prompt_token_cost: float,
    completion_token_cost: float,
    effective_date: Optional[datetime] = datetime.datetime(2024, 9, 17, 21, 4, 51, 971272, tzinfo=datetime.timezone.utc),
    prompt_token_cost_unit: Optional[str] = 'USD',
    completion_token_cost_unit: Optional[str] = 'USD',
    provider_id: Optional[str] = 'default'
) → CostCreateRes

Add a cost to the current project.

Examples:

     client.add_cost(llm_id="my_expensive_custom_model", prompt_token_cost=1, completion_token_cost=2)
     client.add_cost(llm_id="my_expensive_custom_model", prompt_token_cost=500, completion_token_cost=1000, effective_date=datetime(1998, 10, 3))

Args:

  • llm_id: The ID of the LLM. eg "gpt-4o-mini-2024-07-18"
  • prompt_token_cost: The cost of the prompt tokens. eg .0005
  • completion_token_cost: The cost of the completion tokens. eg .0015
  • effective_date: Defaults to the current date. A datetime.datetime object.
  • provider_id: The provider of the LLM. Defaults to "default". eg "openai"
  • prompt_token_cost_unit: The unit of the cost for the prompt tokens. Defaults to "USD". (Currently unused, will be used in the future to specify the currency type for the cost eg "tokens" or "time")
  • completion_token_cost_unit: The unit of the cost for the completion tokens. Defaults to "USD". (Currently unused, will be used in the future to specify the currency type for the cost eg "tokens" or "time")

Returns: A CostCreateRes object. Which has one field called a list of tuples called ids. Each tuple contains the llm_id and the id of the created cost object.

method call​

call(call_id: str, include_costs: Optional[bool] = False) → WeaveObject

method calls​

calls(
    filter: Optional[CallsFilter] = None,
    include_costs: Optional[bool] = False
) → CallsIter

method create_call​

create_call(
    op: Union[str, Op],
    inputs: dict,
    parent: Optional[Call] = None,
    attributes: Optional[dict] = None,
    display_name: Optional[str, Callable[[Call], str]] = None,
    use_stack: bool = True
) → Call

Create, log, and push a call onto the runtime stack.

Args:

  • op: The operation producing the call, or the name of an anonymous operation.
  • inputs: The inputs to the operation.
  • parent: The parent call. If parent is not provided, the current run is used as the parent.
  • display_name: The display name for the call. Defaults to None.
  • attributes: The attributes for the call. Defaults to None.
  • use_stack: Whether to push the call onto the runtime stack. Defaults to True.

Returns: The created Call object.

method delete_call​

delete_call(call: Call) → None

method fail_call​

fail_call(call: Call, exception: BaseException) → None

Fail a call with an exception. This is a convenience method for finish_call.

method feedback​

feedback(
    query: Optional[Query, str] = None,
    reaction: Optional[str] = None,
    offset: int = 0,
    limit: int = 100
) → FeedbackQuery

method finish_call​

finish_call(
    call: Call,
    output: Any = None,
    exception: Optional[BaseException] = None,
    postprocess_output: Optional[Callable[, Any]] = None
) → None

method get​

get(ref: ObjectRef) → Any

method get_call​

get_call(call_id: str, include_costs: Optional[bool] = False) → WeaveObject

method get_calls​

get_calls(
    filter: Optional[CallsFilter] = None,
    include_costs: Optional[bool] = False
) → CallsIter

method get_feedback​

get_feedback(
    query: Optional[Query, str] = None,
    reaction: Optional[str] = None,
    offset: int = 0,
    limit: int = 100
) → FeedbackQuery

Query project for feedback.

Examples:

    # Fetch a specific feedback object.
    # Note that this still returns a collection, which is expected
    # to contain zero or one item(s).
    client.get_feedback("1B4082A3-4EDA-4BEB-BFEB-2D16ED59AA07")

    # Find all feedback objects with a specific reaction.
    client.get_feedback(reaction="👍", limit=10)

Args:

  • query: A mongo-style query expression. For convenience, also accepts a feedback UUID string.
  • reaction: For convenience, filter by a particular reaction emoji.
  • offset: The offset to start fetching feedback objects from.
  • limit: The maximum number of feedback objects to fetch.

Returns: A FeedbackQuery object.

method purge_costs​

purge_costs(ids: Union[list[str], str]) → None

Purge costs from the current project.

Examples:

     client.purge_costs([ids])
     client.purge_costs(ids)

Args:

  • ids: The cost IDs to purge. Can be a single ID or a list of IDs.

method query_costs​

query_costs(
    query: Optional[Query, str] = None,
    llm_ids: Optional[list[str]] = None,
    offset: int = 0,
    limit: int = 100
) → list[CostQueryOutput]

Query project for costs.

Examples:

     # Fetch a specific cost object.
     # Note that this still returns a collection, which is expected
     # to contain zero or one item(s).
     client.query_costs("1B4082A3-4EDA-4BEB-BFEB-2D16ED59AA07")

     # Find all cost objects with a specific reaction.
     client.query_costs(llm_ids=["gpt-4o-mini-2024-07-18"], limit=10)

Args:

  • query: A mongo-style query expression. For convenience, also accepts a cost UUID string.
  • llm_ids: For convenience, filter for a set of llm_ids.
  • offset: The offset to start fetching cost objects from.
  • limit: The maximum number of cost objects to fetch.

Returns: A CostQuery object.

method save​

save(val: Any, name: str, branch: str = 'latest') → Any

class Call​

A Call represents a single operation that was executed as part of a trace.

method __init__​

__init__(
    op_name: str,
    trace_id: str,
    project_id: str,
    parent_id: Optional[str],
    inputs: dict,
    id: Optional[str] = None,
    output: Any = None,
    exception: Optional[str] = None,
    summary: Optional[dict] = None,
    display_name: Optional[str, Callable[[ForwardRef('Call')], str]] = None,
    attributes: Optional[dict] = None,
    started_at: Optional[datetime] = None,
    ended_at: Optional[datetime] = None,
    deleted_at: Optional[datetime] = None,
    _children: list['Call'] = <factory>,
    _feedback: Optional[RefFeedbackQuery] = None
) → None

property feedback​

property func_name​

The decorated function's name that produced this call.

This is different from op_name which is usually the ref of the op.

property ui_url​

method children​

children() → CallsIter

method delete​

delete() → bool

Delete the call.

method remove_display_name​

remove_display_name() → None

method set_display_name​

set_display_name(name: Optional[str]) → None

Set the display name for the call.

Args:

  • name: The display name to set for the call.

Example:

result, call = my_function.call("World")
call.set_display_name("My Custom Display Name")

class CallsIter​

method __init__​

__init__(
    server: TraceServerInterface,
    project_id: str,
    filter: CallsFilter,
    include_costs: bool = False
) → None