# This file was auto-generated by Fern from our API Definition. import datetime as dt import typing from ..core.client_wrapper import AsyncClientWrapper, SyncClientWrapper from ..core.request_options import RequestOptions from ..types.comment import Comment from ..types.error_info import ErrorInfo from ..types.error_info_write import ErrorInfoWrite from ..types.feedback_score_batch_item import FeedbackScoreBatchItem from ..types.feedback_score_batch_item_thread import FeedbackScoreBatchItemThread from ..types.feedback_score_names_public import FeedbackScoreNamesPublic from ..types.feedback_score_source import FeedbackScoreSource from ..types.json_list_string import JsonListString from ..types.json_list_string_write import JsonListStringWrite from ..types.project_stats_public import ProjectStatsPublic from ..types.trace_filter_public import TraceFilterPublic from ..types.trace_page_public import TracePagePublic from ..types.trace_public import TracePublic from ..types.trace_thread import TraceThread from ..types.trace_thread_filter import TraceThreadFilter from ..types.trace_thread_page import TraceThreadPage from ..types.trace_thread_update import TraceThreadUpdate from ..types.trace_update import TraceUpdate from ..types.trace_write import TraceWrite from ..types.value_entry import ValueEntry from .raw_client import AsyncRawTracesClient, RawTracesClient # this is used as the default value for optional parameters OMIT = typing.cast(typing.Any, ...) class TracesClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._raw_client = RawTracesClient(client_wrapper=client_wrapper) @property def with_raw_response(self) -> RawTracesClient: """ Retrieves a raw implementation of this client that returns raw responses. Returns ------- RawTracesClient """ return self._raw_client def add_thread_comment( self, id_: str, *, text: str, id: typing.Optional[str] = OMIT, created_at: typing.Optional[dt.datetime] = OMIT, last_updated_at: typing.Optional[dt.datetime] = OMIT, created_by: typing.Optional[str] = OMIT, last_updated_by: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Add thread comment Parameters ---------- id_ : str text : str id : typing.Optional[str] created_at : typing.Optional[dt.datetime] last_updated_at : typing.Optional[dt.datetime] created_by : typing.Optional[str] last_updated_by : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.traces.add_thread_comment(id_='id', text='text', ) """ _response = self._raw_client.add_thread_comment( id_, text=text, id=id, created_at=created_at, last_updated_at=last_updated_at, created_by=created_by, last_updated_by=last_updated_by, request_options=request_options, ) return _response.data def add_trace_comment( self, id_: str, *, text: str, id: typing.Optional[str] = OMIT, created_at: typing.Optional[dt.datetime] = OMIT, last_updated_at: typing.Optional[dt.datetime] = OMIT, created_by: typing.Optional[str] = OMIT, last_updated_by: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Add trace comment Parameters ---------- id_ : str text : str id : typing.Optional[str] created_at : typing.Optional[dt.datetime] last_updated_at : typing.Optional[dt.datetime] created_by : typing.Optional[str] last_updated_by : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.traces.add_trace_comment(id_='id', text='text', ) """ _response = self._raw_client.add_trace_comment( id_, text=text, id=id, created_at=created_at, last_updated_at=last_updated_at, created_by=created_by, last_updated_by=last_updated_by, request_options=request_options, ) return _response.data def add_trace_feedback_score( self, id: str, *, name: str, value: float, source: FeedbackScoreSource, category_name: typing.Optional[str] = OMIT, reason: typing.Optional[str] = OMIT, created_at: typing.Optional[dt.datetime] = OMIT, last_updated_at: typing.Optional[dt.datetime] = OMIT, created_by: typing.Optional[str] = OMIT, last_updated_by: typing.Optional[str] = OMIT, value_by_author: typing.Optional[typing.Dict[str, ValueEntry]] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Add trace feedback score Parameters ---------- id : str name : str value : float source : FeedbackScoreSource category_name : typing.Optional[str] reason : typing.Optional[str] created_at : typing.Optional[dt.datetime] last_updated_at : typing.Optional[dt.datetime] created_by : typing.Optional[str] last_updated_by : typing.Optional[str] value_by_author : typing.Optional[typing.Dict[str, ValueEntry]] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.traces.add_trace_feedback_score(id='id', name='name', value=1.1, source="ui", ) """ _response = self._raw_client.add_trace_feedback_score( id, name=name, value=value, source=source, category_name=category_name, reason=reason, created_at=created_at, last_updated_at=last_updated_at, created_by=created_by, last_updated_by=last_updated_by, value_by_author=value_by_author, request_options=request_options, ) return _response.data def create_traces( self, *, traces: typing.Sequence[TraceWrite], request_options: typing.Optional[RequestOptions] = None ) -> None: """ Create traces Parameters ---------- traces : typing.Sequence[TraceWrite] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import OpikApi from Opik import TraceWrite import datetime client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.traces.create_traces(traces=[TraceWrite(start_time=datetime.datetime.fromisoformat("2024-01-15 09:30:00+00:00", ), )], ) """ _response = self._raw_client.create_traces(traces=traces, request_options=request_options) return _response.data def batch_update_traces( self, *, ids: typing.Sequence[str], update: TraceUpdate, merge_tags: typing.Optional[bool] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Update multiple traces Parameters ---------- ids : typing.Sequence[str] List of trace IDs to update (max 1000) update : TraceUpdate merge_tags : typing.Optional[bool] If true, merge tags with existing tags instead of replacing them. Default: false request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import OpikApi from Opik import TraceUpdate client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.traces.batch_update_traces(ids=['ids'], update=TraceUpdate(), ) """ _response = self._raw_client.batch_update_traces( ids=ids, update=update, merge_tags=merge_tags, request_options=request_options ) return _response.data def batch_update_threads( self, *, ids: typing.Sequence[str], update: TraceThreadUpdate, merge_tags: typing.Optional[bool] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Update multiple threads Parameters ---------- ids : typing.Sequence[str] List of thread model IDs to update (max 1000) update : TraceThreadUpdate merge_tags : typing.Optional[bool] If true, merge tags with existing tags instead of replacing them. Default: false request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import OpikApi from Opik import TraceThreadUpdate client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.traces.batch_update_threads(ids=['ids'], update=TraceThreadUpdate(), ) """ _response = self._raw_client.batch_update_threads( ids=ids, update=update, merge_tags=merge_tags, request_options=request_options ) return _response.data def close_trace_thread( self, *, project_name: typing.Optional[str] = OMIT, project_id: typing.Optional[str] = OMIT, thread_id: typing.Optional[str] = OMIT, thread_ids: typing.Optional[typing.Sequence[str]] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Close one or multiple trace threads. Supports both single thread_id and multiple thread_ids for batch operations. Parameters ---------- project_name : typing.Optional[str] project_id : typing.Optional[str] thread_id : typing.Optional[str] thread_ids : typing.Optional[typing.Sequence[str]] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.traces.close_trace_thread() """ _response = self._raw_client.close_trace_thread( project_name=project_name, project_id=project_id, thread_id=thread_id, thread_ids=thread_ids, request_options=request_options, ) return _response.data def get_traces_by_project( self, *, page: typing.Optional[int] = None, size: typing.Optional[int] = None, project_name: typing.Optional[str] = None, project_id: typing.Optional[str] = None, filters: typing.Optional[str] = None, truncate: typing.Optional[bool] = None, strip_attachments: typing.Optional[bool] = None, sorting: typing.Optional[str] = None, exclude: typing.Optional[str] = None, search: typing.Optional[str] = None, from_time: typing.Optional[dt.datetime] = None, to_time: typing.Optional[dt.datetime] = None, request_options: typing.Optional[RequestOptions] = None, ) -> TracePagePublic: """ Get traces by project_name or project_id Parameters ---------- page : typing.Optional[int] size : typing.Optional[int] project_name : typing.Optional[str] project_id : typing.Optional[str] filters : typing.Optional[str] truncate : typing.Optional[bool] strip_attachments : typing.Optional[bool] sorting : typing.Optional[str] exclude : typing.Optional[str] search : typing.Optional[str] from_time : typing.Optional[dt.datetime] to_time : typing.Optional[dt.datetime] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- TracePagePublic Trace resource Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.traces.get_traces_by_project() """ _response = self._raw_client.get_traces_by_project( page=page, size=size, project_name=project_name, project_id=project_id, filters=filters, truncate=truncate, strip_attachments=strip_attachments, sorting=sorting, exclude=exclude, search=search, from_time=from_time, to_time=to_time, request_options=request_options, ) return _response.data def create_trace( self, *, start_time: dt.datetime, id: typing.Optional[str] = OMIT, project_name: typing.Optional[str] = OMIT, name: typing.Optional[str] = OMIT, end_time: typing.Optional[dt.datetime] = OMIT, input: typing.Optional[JsonListStringWrite] = OMIT, output: typing.Optional[JsonListStringWrite] = OMIT, metadata: typing.Optional[JsonListStringWrite] = OMIT, tags: typing.Optional[typing.Sequence[str]] = OMIT, error_info: typing.Optional[ErrorInfoWrite] = OMIT, last_updated_at: typing.Optional[dt.datetime] = OMIT, ttft: typing.Optional[float] = OMIT, thread_id: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Get trace Parameters ---------- start_time : dt.datetime id : typing.Optional[str] project_name : typing.Optional[str] If null, the default project is used name : typing.Optional[str] end_time : typing.Optional[dt.datetime] input : typing.Optional[JsonListStringWrite] output : typing.Optional[JsonListStringWrite] metadata : typing.Optional[JsonListStringWrite] tags : typing.Optional[typing.Sequence[str]] error_info : typing.Optional[ErrorInfoWrite] last_updated_at : typing.Optional[dt.datetime] ttft : typing.Optional[float] Time to first token in milliseconds thread_id : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import OpikApi import datetime client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.traces.create_trace(start_time=datetime.datetime.fromisoformat("2024-01-15 09:30:00+00:00", ), ) """ _response = self._raw_client.create_trace( start_time=start_time, id=id, project_name=project_name, name=name, end_time=end_time, input=input, output=output, metadata=metadata, tags=tags, error_info=error_info, last_updated_at=last_updated_at, ttft=ttft, thread_id=thread_id, request_options=request_options, ) return _response.data def get_trace_by_id( self, id: str, *, strip_attachments: typing.Optional[bool] = None, request_options: typing.Optional[RequestOptions] = None, ) -> TracePublic: """ Get trace by id Parameters ---------- id : str strip_attachments : typing.Optional[bool] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- TracePublic Trace resource Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.traces.get_trace_by_id(id='id', ) """ _response = self._raw_client.get_trace_by_id( id, strip_attachments=strip_attachments, request_options=request_options ) return _response.data def delete_trace_by_id(self, id: str, *, request_options: typing.Optional[RequestOptions] = None) -> None: """ Delete trace by id Parameters ---------- id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.traces.delete_trace_by_id(id='id', ) """ _response = self._raw_client.delete_trace_by_id(id, request_options=request_options) return _response.data def update_trace( self, id: str, *, project_name: typing.Optional[str] = OMIT, project_id: typing.Optional[str] = OMIT, name: typing.Optional[str] = OMIT, end_time: typing.Optional[dt.datetime] = OMIT, input: typing.Optional[JsonListString] = OMIT, output: typing.Optional[JsonListString] = OMIT, metadata: typing.Optional[JsonListString] = OMIT, tags: typing.Optional[typing.Sequence[str]] = OMIT, tags_to_add: typing.Optional[typing.Sequence[str]] = OMIT, tags_to_remove: typing.Optional[typing.Sequence[str]] = OMIT, error_info: typing.Optional[ErrorInfo] = OMIT, thread_id: typing.Optional[str] = OMIT, ttft: typing.Optional[float] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Update trace by id Parameters ---------- id : str project_name : typing.Optional[str] If null and project_id not specified, Default Project is assumed project_id : typing.Optional[str] If null and project_name not specified, Default Project is assumed name : typing.Optional[str] end_time : typing.Optional[dt.datetime] input : typing.Optional[JsonListString] output : typing.Optional[JsonListString] metadata : typing.Optional[JsonListString] tags : typing.Optional[typing.Sequence[str]] Tags tags_to_add : typing.Optional[typing.Sequence[str]] Tags to add tags_to_remove : typing.Optional[typing.Sequence[str]] Tags to remove error_info : typing.Optional[ErrorInfo] thread_id : typing.Optional[str] ttft : typing.Optional[float] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.traces.update_trace(id='id', ) """ _response = self._raw_client.update_trace( id, project_name=project_name, project_id=project_id, name=name, end_time=end_time, input=input, output=output, metadata=metadata, tags=tags, tags_to_add=tags_to_add, tags_to_remove=tags_to_remove, error_info=error_info, thread_id=thread_id, ttft=ttft, request_options=request_options, ) return _response.data def delete_thread_comments( self, *, ids: typing.Sequence[str], request_options: typing.Optional[RequestOptions] = None ) -> None: """ Delete thread comments Parameters ---------- ids : typing.Sequence[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.traces.delete_thread_comments(ids=['ids'], ) """ _response = self._raw_client.delete_thread_comments(ids=ids, request_options=request_options) return _response.data def delete_thread_feedback_scores( self, *, project_name: str, thread_id: str, names: typing.Sequence[str], author: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Delete thread feedback scores Parameters ---------- project_name : str thread_id : str names : typing.Sequence[str] author : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.traces.delete_thread_feedback_scores(project_name='project_name', thread_id='thread_id', names=['names'], ) """ _response = self._raw_client.delete_thread_feedback_scores( project_name=project_name, thread_id=thread_id, names=names, author=author, request_options=request_options ) return _response.data def delete_trace_comments( self, *, ids: typing.Sequence[str], request_options: typing.Optional[RequestOptions] = None ) -> None: """ Delete trace comments Parameters ---------- ids : typing.Sequence[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.traces.delete_trace_comments(ids=['ids'], ) """ _response = self._raw_client.delete_trace_comments(ids=ids, request_options=request_options) return _response.data def delete_trace_feedback_score( self, id: str, *, name: str, author: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Delete trace feedback score Parameters ---------- id : str name : str author : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.traces.delete_trace_feedback_score(id='id', name='name', ) """ _response = self._raw_client.delete_trace_feedback_score( id, name=name, author=author, request_options=request_options ) return _response.data def delete_trace_threads( self, *, thread_ids: typing.Sequence[str], project_name: typing.Optional[str] = OMIT, project_id: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Delete trace threads Parameters ---------- thread_ids : typing.Sequence[str] project_name : typing.Optional[str] If null, project_id must be provided project_id : typing.Optional[str] If null, project_name must be provided request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.traces.delete_trace_threads(thread_ids=['thread_ids'], ) """ _response = self._raw_client.delete_trace_threads( thread_ids=thread_ids, project_name=project_name, project_id=project_id, request_options=request_options ) return _response.data def delete_traces( self, *, ids: typing.Sequence[str], request_options: typing.Optional[RequestOptions] = None ) -> None: """ Delete traces Parameters ---------- ids : typing.Sequence[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.traces.delete_traces(ids=['ids'], ) """ _response = self._raw_client.delete_traces(ids=ids, request_options=request_options) return _response.data def find_feedback_score_names2( self, *, project_id: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None ) -> FeedbackScoreNamesPublic: """ Find Feedback Score names Parameters ---------- project_id : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- FeedbackScoreNamesPublic Feedback Scores resource Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.traces.find_feedback_score_names2() """ _response = self._raw_client.find_feedback_score_names2(project_id=project_id, request_options=request_options) return _response.data def find_trace_threads_feedback_score_names( self, *, project_id: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None ) -> FeedbackScoreNamesPublic: """ Find Trace Threads Feedback Score names Parameters ---------- project_id : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- FeedbackScoreNamesPublic Find Trace Threads Feedback Score names Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.traces.find_trace_threads_feedback_score_names() """ _response = self._raw_client.find_trace_threads_feedback_score_names( project_id=project_id, request_options=request_options ) return _response.data def get_trace_stats( self, *, project_id: typing.Optional[str] = None, project_name: typing.Optional[str] = None, filters: typing.Optional[str] = None, search: typing.Optional[str] = None, from_time: typing.Optional[dt.datetime] = None, to_time: typing.Optional[dt.datetime] = None, request_options: typing.Optional[RequestOptions] = None, ) -> ProjectStatsPublic: """ Get trace stats Parameters ---------- project_id : typing.Optional[str] project_name : typing.Optional[str] filters : typing.Optional[str] search : typing.Optional[str] from_time : typing.Optional[dt.datetime] to_time : typing.Optional[dt.datetime] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- ProjectStatsPublic Trace stats resource Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.traces.get_trace_stats() """ _response = self._raw_client.get_trace_stats( project_id=project_id, project_name=project_name, filters=filters, search=search, from_time=from_time, to_time=to_time, request_options=request_options, ) return _response.data def get_thread_comment( self, thread_id: str, comment_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> Comment: """ Get thread comment Parameters ---------- thread_id : str comment_id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- Comment Comment resource Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.traces.get_thread_comment(thread_id='threadId', comment_id='commentId', ) """ _response = self._raw_client.get_thread_comment(thread_id, comment_id, request_options=request_options) return _response.data def get_trace_thread_stats( self, *, project_id: typing.Optional[str] = None, project_name: typing.Optional[str] = None, filters: typing.Optional[str] = None, search: typing.Optional[str] = None, from_time: typing.Optional[dt.datetime] = None, to_time: typing.Optional[dt.datetime] = None, request_options: typing.Optional[RequestOptions] = None, ) -> ProjectStatsPublic: """ Get trace thread stats Parameters ---------- project_id : typing.Optional[str] project_name : typing.Optional[str] filters : typing.Optional[str] search : typing.Optional[str] from_time : typing.Optional[dt.datetime] to_time : typing.Optional[dt.datetime] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- ProjectStatsPublic Trace thread stats resource Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.traces.get_trace_thread_stats() """ _response = self._raw_client.get_trace_thread_stats( project_id=project_id, project_name=project_name, filters=filters, search=search, from_time=from_time, to_time=to_time, request_options=request_options, ) return _response.data def get_trace_comment( self, trace_id: str, comment_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> Comment: """ Get trace comment Parameters ---------- trace_id : str comment_id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- Comment Comment resource Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.traces.get_trace_comment(trace_id='traceId', comment_id='commentId', ) """ _response = self._raw_client.get_trace_comment(trace_id, comment_id, request_options=request_options) return _response.data def get_trace_thread( self, *, thread_id: str, project_name: typing.Optional[str] = OMIT, project_id: typing.Optional[str] = OMIT, truncate: typing.Optional[bool] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> TraceThread: """ Get trace thread Parameters ---------- thread_id : str project_name : typing.Optional[str] project_id : typing.Optional[str] truncate : typing.Optional[bool] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- TraceThread Trace thread resource Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.traces.get_trace_thread(thread_id='thread_id', ) """ _response = self._raw_client.get_trace_thread( thread_id=thread_id, project_name=project_name, project_id=project_id, truncate=truncate, request_options=request_options, ) return _response.data def get_trace_threads( self, *, page: typing.Optional[int] = None, size: typing.Optional[int] = None, project_name: typing.Optional[str] = None, project_id: typing.Optional[str] = None, truncate: typing.Optional[bool] = None, strip_attachments: typing.Optional[bool] = None, filters: typing.Optional[str] = None, sorting: typing.Optional[str] = None, search: typing.Optional[str] = None, from_time: typing.Optional[dt.datetime] = None, to_time: typing.Optional[dt.datetime] = None, request_options: typing.Optional[RequestOptions] = None, ) -> TraceThreadPage: """ Get trace threads Parameters ---------- page : typing.Optional[int] size : typing.Optional[int] project_name : typing.Optional[str] project_id : typing.Optional[str] truncate : typing.Optional[bool] strip_attachments : typing.Optional[bool] filters : typing.Optional[str] sorting : typing.Optional[str] search : typing.Optional[str] from_time : typing.Optional[dt.datetime] to_time : typing.Optional[dt.datetime] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- TraceThreadPage Trace threads resource Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.traces.get_trace_threads() """ _response = self._raw_client.get_trace_threads( page=page, size=size, project_name=project_name, project_id=project_id, truncate=truncate, strip_attachments=strip_attachments, filters=filters, sorting=sorting, search=search, from_time=from_time, to_time=to_time, request_options=request_options, ) return _response.data def open_trace_thread( self, *, thread_id: str, project_name: typing.Optional[str] = OMIT, project_id: typing.Optional[str] = OMIT, truncate: typing.Optional[bool] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Open trace thread Parameters ---------- thread_id : str project_name : typing.Optional[str] project_id : typing.Optional[str] truncate : typing.Optional[bool] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.traces.open_trace_thread(thread_id='thread_id', ) """ _response = self._raw_client.open_trace_thread( thread_id=thread_id, project_name=project_name, project_id=project_id, truncate=truncate, request_options=request_options, ) return _response.data def score_batch_of_threads( self, *, scores: typing.Sequence[FeedbackScoreBatchItemThread], request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Batch feedback scoring for threads Parameters ---------- scores : typing.Sequence[FeedbackScoreBatchItemThread] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import OpikApi from Opik import FeedbackScoreBatchItemThread client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.traces.score_batch_of_threads(scores=[FeedbackScoreBatchItemThread(name='name', value=1.1, source="ui", thread_id='thread_id', )], ) """ _response = self._raw_client.score_batch_of_threads(scores=scores, request_options=request_options) return _response.data def score_batch_of_traces( self, *, scores: typing.Sequence[FeedbackScoreBatchItem], request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Batch feedback scoring for traces Parameters ---------- scores : typing.Sequence[FeedbackScoreBatchItem] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import OpikApi from Opik import FeedbackScoreBatchItem client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.traces.score_batch_of_traces(scores=[FeedbackScoreBatchItem(name='name', value=1.1, source="ui", id='id', )], ) """ _response = self._raw_client.score_batch_of_traces(scores=scores, request_options=request_options) return _response.data def search_trace_threads( self, *, project_name: typing.Optional[str] = OMIT, project_id: typing.Optional[str] = OMIT, filters: typing.Optional[typing.Sequence[TraceThreadFilter]] = OMIT, last_retrieved_thread_model_id: typing.Optional[str] = OMIT, limit: typing.Optional[int] = OMIT, truncate: typing.Optional[bool] = OMIT, strip_attachments: typing.Optional[bool] = OMIT, from_time: typing.Optional[dt.datetime] = OMIT, to_time: typing.Optional[dt.datetime] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> typing.Iterator[bytes]: """ Search trace threads Parameters ---------- project_name : typing.Optional[str] project_id : typing.Optional[str] filters : typing.Optional[typing.Sequence[TraceThreadFilter]] last_retrieved_thread_model_id : typing.Optional[str] limit : typing.Optional[int] Max number of trace thread to be streamed truncate : typing.Optional[bool] Truncate input, output and metadata to slim payloads strip_attachments : typing.Optional[bool] If true, returns attachment references like [file.png]; if false, downloads and reinjects stripped attachments from_time : typing.Optional[dt.datetime] Filter trace threads created from this time (ISO-8601 format). to_time : typing.Optional[dt.datetime] Filter trace threads created up to this time (ISO-8601 format). If not provided, defaults to current time. Must be after 'from_time'. request_options : typing.Optional[RequestOptions] Request-specific configuration. You can pass in configuration such as `chunk_size`, and more to customize the request and response. Returns ------- typing.Iterator[bytes] Trace threads stream or error during process Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.traces.search_trace_threads() """ with self._raw_client.search_trace_threads( project_name=project_name, project_id=project_id, filters=filters, last_retrieved_thread_model_id=last_retrieved_thread_model_id, limit=limit, truncate=truncate, strip_attachments=strip_attachments, from_time=from_time, to_time=to_time, request_options=request_options, ) as r: yield from r.data def search_traces( self, *, project_name: typing.Optional[str] = OMIT, project_id: typing.Optional[str] = OMIT, filters: typing.Optional[typing.Sequence[TraceFilterPublic]] = OMIT, last_retrieved_id: typing.Optional[str] = OMIT, limit: typing.Optional[int] = OMIT, truncate: typing.Optional[bool] = OMIT, strip_attachments: typing.Optional[bool] = OMIT, from_time: typing.Optional[dt.datetime] = OMIT, to_time: typing.Optional[dt.datetime] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> typing.Iterator[bytes]: """ Search traces Parameters ---------- project_name : typing.Optional[str] project_id : typing.Optional[str] filters : typing.Optional[typing.Sequence[TraceFilterPublic]] last_retrieved_id : typing.Optional[str] limit : typing.Optional[int] Max number of traces to be streamed truncate : typing.Optional[bool] Truncate input, output and metadata to slim payloads strip_attachments : typing.Optional[bool] If true, returns attachment references like [file.png]; if false, downloads and reinjects stripped attachments from_time : typing.Optional[dt.datetime] Filter traces created from this time (ISO-8601 format). to_time : typing.Optional[dt.datetime] Filter traces created up to this time (ISO-8601 format). If not provided, defaults to current time. Must be after 'from_time'. request_options : typing.Optional[RequestOptions] Request-specific configuration. You can pass in configuration such as `chunk_size`, and more to customize the request and response. Returns ------- typing.Iterator[bytes] Traces stream or error during process Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.traces.search_traces() """ with self._raw_client.search_traces( project_name=project_name, project_id=project_id, filters=filters, last_retrieved_id=last_retrieved_id, limit=limit, truncate=truncate, strip_attachments=strip_attachments, from_time=from_time, to_time=to_time, request_options=request_options, ) as r: yield from r.data def update_thread( self, thread_model_id: str, *, tags: typing.Optional[typing.Sequence[str]] = OMIT, tags_to_add: typing.Optional[typing.Sequence[str]] = OMIT, tags_to_remove: typing.Optional[typing.Sequence[str]] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Update thread Parameters ---------- thread_model_id : str tags : typing.Optional[typing.Sequence[str]] tags_to_add : typing.Optional[typing.Sequence[str]] tags_to_remove : typing.Optional[typing.Sequence[str]] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.traces.update_thread(thread_model_id='threadModelId', ) """ _response = self._raw_client.update_thread( thread_model_id, tags=tags, tags_to_add=tags_to_add, tags_to_remove=tags_to_remove, request_options=request_options, ) return _response.data def update_thread_comment( self, comment_id: str, *, text: str, id: typing.Optional[str] = OMIT, created_at: typing.Optional[dt.datetime] = OMIT, last_updated_at: typing.Optional[dt.datetime] = OMIT, created_by: typing.Optional[str] = OMIT, last_updated_by: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Update thread comment by id Parameters ---------- comment_id : str text : str id : typing.Optional[str] created_at : typing.Optional[dt.datetime] last_updated_at : typing.Optional[dt.datetime] created_by : typing.Optional[str] last_updated_by : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.traces.update_thread_comment(comment_id='commentId', text='text', ) """ _response = self._raw_client.update_thread_comment( comment_id, text=text, id=id, created_at=created_at, last_updated_at=last_updated_at, created_by=created_by, last_updated_by=last_updated_by, request_options=request_options, ) return _response.data def update_trace_comment( self, comment_id: str, *, text: str, id: typing.Optional[str] = OMIT, created_at: typing.Optional[dt.datetime] = OMIT, last_updated_at: typing.Optional[dt.datetime] = OMIT, created_by: typing.Optional[str] = OMIT, last_updated_by: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Update trace comment by id Parameters ---------- comment_id : str text : str id : typing.Optional[str] created_at : typing.Optional[dt.datetime] last_updated_at : typing.Optional[dt.datetime] created_by : typing.Optional[str] last_updated_by : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.traces.update_trace_comment(comment_id='commentId', text='text', ) """ _response = self._raw_client.update_trace_comment( comment_id, text=text, id=id, created_at=created_at, last_updated_at=last_updated_at, created_by=created_by, last_updated_by=last_updated_by, request_options=request_options, ) return _response.data class AsyncTracesClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._raw_client = AsyncRawTracesClient(client_wrapper=client_wrapper) @property def with_raw_response(self) -> AsyncRawTracesClient: """ Retrieves a raw implementation of this client that returns raw responses. Returns ------- AsyncRawTracesClient """ return self._raw_client async def add_thread_comment( self, id_: str, *, text: str, id: typing.Optional[str] = OMIT, created_at: typing.Optional[dt.datetime] = OMIT, last_updated_at: typing.Optional[dt.datetime] = OMIT, created_by: typing.Optional[str] = OMIT, last_updated_by: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Add thread comment Parameters ---------- id_ : str text : str id : typing.Optional[str] created_at : typing.Optional[dt.datetime] last_updated_at : typing.Optional[dt.datetime] created_by : typing.Optional[str] last_updated_by : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.traces.add_thread_comment(id_='id', text='text', ) asyncio.run(main()) """ _response = await self._raw_client.add_thread_comment( id_, text=text, id=id, created_at=created_at, last_updated_at=last_updated_at, created_by=created_by, last_updated_by=last_updated_by, request_options=request_options, ) return _response.data async def add_trace_comment( self, id_: str, *, text: str, id: typing.Optional[str] = OMIT, created_at: typing.Optional[dt.datetime] = OMIT, last_updated_at: typing.Optional[dt.datetime] = OMIT, created_by: typing.Optional[str] = OMIT, last_updated_by: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Add trace comment Parameters ---------- id_ : str text : str id : typing.Optional[str] created_at : typing.Optional[dt.datetime] last_updated_at : typing.Optional[dt.datetime] created_by : typing.Optional[str] last_updated_by : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.traces.add_trace_comment(id_='id', text='text', ) asyncio.run(main()) """ _response = await self._raw_client.add_trace_comment( id_, text=text, id=id, created_at=created_at, last_updated_at=last_updated_at, created_by=created_by, last_updated_by=last_updated_by, request_options=request_options, ) return _response.data async def add_trace_feedback_score( self, id: str, *, name: str, value: float, source: FeedbackScoreSource, category_name: typing.Optional[str] = OMIT, reason: typing.Optional[str] = OMIT, created_at: typing.Optional[dt.datetime] = OMIT, last_updated_at: typing.Optional[dt.datetime] = OMIT, created_by: typing.Optional[str] = OMIT, last_updated_by: typing.Optional[str] = OMIT, value_by_author: typing.Optional[typing.Dict[str, ValueEntry]] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Add trace feedback score Parameters ---------- id : str name : str value : float source : FeedbackScoreSource category_name : typing.Optional[str] reason : typing.Optional[str] created_at : typing.Optional[dt.datetime] last_updated_at : typing.Optional[dt.datetime] created_by : typing.Optional[str] last_updated_by : typing.Optional[str] value_by_author : typing.Optional[typing.Dict[str, ValueEntry]] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.traces.add_trace_feedback_score(id='id', name='name', value=1.1, source="ui", ) asyncio.run(main()) """ _response = await self._raw_client.add_trace_feedback_score( id, name=name, value=value, source=source, category_name=category_name, reason=reason, created_at=created_at, last_updated_at=last_updated_at, created_by=created_by, last_updated_by=last_updated_by, value_by_author=value_by_author, request_options=request_options, ) return _response.data async def create_traces( self, *, traces: typing.Sequence[TraceWrite], request_options: typing.Optional[RequestOptions] = None ) -> None: """ Create traces Parameters ---------- traces : typing.Sequence[TraceWrite] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import AsyncOpikApi from Opik import TraceWrite import datetime import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.traces.create_traces(traces=[TraceWrite(start_time=datetime.datetime.fromisoformat("2024-01-15 09:30:00+00:00", ), )], ) asyncio.run(main()) """ _response = await self._raw_client.create_traces(traces=traces, request_options=request_options) return _response.data async def batch_update_traces( self, *, ids: typing.Sequence[str], update: TraceUpdate, merge_tags: typing.Optional[bool] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Update multiple traces Parameters ---------- ids : typing.Sequence[str] List of trace IDs to update (max 1000) update : TraceUpdate merge_tags : typing.Optional[bool] If true, merge tags with existing tags instead of replacing them. Default: false request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import AsyncOpikApi from Opik import TraceUpdate import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.traces.batch_update_traces(ids=['ids'], update=TraceUpdate(), ) asyncio.run(main()) """ _response = await self._raw_client.batch_update_traces( ids=ids, update=update, merge_tags=merge_tags, request_options=request_options ) return _response.data async def batch_update_threads( self, *, ids: typing.Sequence[str], update: TraceThreadUpdate, merge_tags: typing.Optional[bool] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Update multiple threads Parameters ---------- ids : typing.Sequence[str] List of thread model IDs to update (max 1000) update : TraceThreadUpdate merge_tags : typing.Optional[bool] If true, merge tags with existing tags instead of replacing them. Default: false request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import AsyncOpikApi from Opik import TraceThreadUpdate import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.traces.batch_update_threads(ids=['ids'], update=TraceThreadUpdate(), ) asyncio.run(main()) """ _response = await self._raw_client.batch_update_threads( ids=ids, update=update, merge_tags=merge_tags, request_options=request_options ) return _response.data async def close_trace_thread( self, *, project_name: typing.Optional[str] = OMIT, project_id: typing.Optional[str] = OMIT, thread_id: typing.Optional[str] = OMIT, thread_ids: typing.Optional[typing.Sequence[str]] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Close one or multiple trace threads. Supports both single thread_id and multiple thread_ids for batch operations. Parameters ---------- project_name : typing.Optional[str] project_id : typing.Optional[str] thread_id : typing.Optional[str] thread_ids : typing.Optional[typing.Sequence[str]] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.traces.close_trace_thread() asyncio.run(main()) """ _response = await self._raw_client.close_trace_thread( project_name=project_name, project_id=project_id, thread_id=thread_id, thread_ids=thread_ids, request_options=request_options, ) return _response.data async def get_traces_by_project( self, *, page: typing.Optional[int] = None, size: typing.Optional[int] = None, project_name: typing.Optional[str] = None, project_id: typing.Optional[str] = None, filters: typing.Optional[str] = None, truncate: typing.Optional[bool] = None, strip_attachments: typing.Optional[bool] = None, sorting: typing.Optional[str] = None, exclude: typing.Optional[str] = None, search: typing.Optional[str] = None, from_time: typing.Optional[dt.datetime] = None, to_time: typing.Optional[dt.datetime] = None, request_options: typing.Optional[RequestOptions] = None, ) -> TracePagePublic: """ Get traces by project_name or project_id Parameters ---------- page : typing.Optional[int] size : typing.Optional[int] project_name : typing.Optional[str] project_id : typing.Optional[str] filters : typing.Optional[str] truncate : typing.Optional[bool] strip_attachments : typing.Optional[bool] sorting : typing.Optional[str] exclude : typing.Optional[str] search : typing.Optional[str] from_time : typing.Optional[dt.datetime] to_time : typing.Optional[dt.datetime] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- TracePagePublic Trace resource Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.traces.get_traces_by_project() asyncio.run(main()) """ _response = await self._raw_client.get_traces_by_project( page=page, size=size, project_name=project_name, project_id=project_id, filters=filters, truncate=truncate, strip_attachments=strip_attachments, sorting=sorting, exclude=exclude, search=search, from_time=from_time, to_time=to_time, request_options=request_options, ) return _response.data async def create_trace( self, *, start_time: dt.datetime, id: typing.Optional[str] = OMIT, project_name: typing.Optional[str] = OMIT, name: typing.Optional[str] = OMIT, end_time: typing.Optional[dt.datetime] = OMIT, input: typing.Optional[JsonListStringWrite] = OMIT, output: typing.Optional[JsonListStringWrite] = OMIT, metadata: typing.Optional[JsonListStringWrite] = OMIT, tags: typing.Optional[typing.Sequence[str]] = OMIT, error_info: typing.Optional[ErrorInfoWrite] = OMIT, last_updated_at: typing.Optional[dt.datetime] = OMIT, ttft: typing.Optional[float] = OMIT, thread_id: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Get trace Parameters ---------- start_time : dt.datetime id : typing.Optional[str] project_name : typing.Optional[str] If null, the default project is used name : typing.Optional[str] end_time : typing.Optional[dt.datetime] input : typing.Optional[JsonListStringWrite] output : typing.Optional[JsonListStringWrite] metadata : typing.Optional[JsonListStringWrite] tags : typing.Optional[typing.Sequence[str]] error_info : typing.Optional[ErrorInfoWrite] last_updated_at : typing.Optional[dt.datetime] ttft : typing.Optional[float] Time to first token in milliseconds thread_id : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import AsyncOpikApi import datetime import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.traces.create_trace(start_time=datetime.datetime.fromisoformat("2024-01-15 09:30:00+00:00", ), ) asyncio.run(main()) """ _response = await self._raw_client.create_trace( start_time=start_time, id=id, project_name=project_name, name=name, end_time=end_time, input=input, output=output, metadata=metadata, tags=tags, error_info=error_info, last_updated_at=last_updated_at, ttft=ttft, thread_id=thread_id, request_options=request_options, ) return _response.data async def get_trace_by_id( self, id: str, *, strip_attachments: typing.Optional[bool] = None, request_options: typing.Optional[RequestOptions] = None, ) -> TracePublic: """ Get trace by id Parameters ---------- id : str strip_attachments : typing.Optional[bool] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- TracePublic Trace resource Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.traces.get_trace_by_id(id='id', ) asyncio.run(main()) """ _response = await self._raw_client.get_trace_by_id( id, strip_attachments=strip_attachments, request_options=request_options ) return _response.data async def delete_trace_by_id(self, id: str, *, request_options: typing.Optional[RequestOptions] = None) -> None: """ Delete trace by id Parameters ---------- id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.traces.delete_trace_by_id(id='id', ) asyncio.run(main()) """ _response = await self._raw_client.delete_trace_by_id(id, request_options=request_options) return _response.data async def update_trace( self, id: str, *, project_name: typing.Optional[str] = OMIT, project_id: typing.Optional[str] = OMIT, name: typing.Optional[str] = OMIT, end_time: typing.Optional[dt.datetime] = OMIT, input: typing.Optional[JsonListString] = OMIT, output: typing.Optional[JsonListString] = OMIT, metadata: typing.Optional[JsonListString] = OMIT, tags: typing.Optional[typing.Sequence[str]] = OMIT, tags_to_add: typing.Optional[typing.Sequence[str]] = OMIT, tags_to_remove: typing.Optional[typing.Sequence[str]] = OMIT, error_info: typing.Optional[ErrorInfo] = OMIT, thread_id: typing.Optional[str] = OMIT, ttft: typing.Optional[float] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Update trace by id Parameters ---------- id : str project_name : typing.Optional[str] If null and project_id not specified, Default Project is assumed project_id : typing.Optional[str] If null and project_name not specified, Default Project is assumed name : typing.Optional[str] end_time : typing.Optional[dt.datetime] input : typing.Optional[JsonListString] output : typing.Optional[JsonListString] metadata : typing.Optional[JsonListString] tags : typing.Optional[typing.Sequence[str]] Tags tags_to_add : typing.Optional[typing.Sequence[str]] Tags to add tags_to_remove : typing.Optional[typing.Sequence[str]] Tags to remove error_info : typing.Optional[ErrorInfo] thread_id : typing.Optional[str] ttft : typing.Optional[float] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.traces.update_trace(id='id', ) asyncio.run(main()) """ _response = await self._raw_client.update_trace( id, project_name=project_name, project_id=project_id, name=name, end_time=end_time, input=input, output=output, metadata=metadata, tags=tags, tags_to_add=tags_to_add, tags_to_remove=tags_to_remove, error_info=error_info, thread_id=thread_id, ttft=ttft, request_options=request_options, ) return _response.data async def delete_thread_comments( self, *, ids: typing.Sequence[str], request_options: typing.Optional[RequestOptions] = None ) -> None: """ Delete thread comments Parameters ---------- ids : typing.Sequence[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.traces.delete_thread_comments(ids=['ids'], ) asyncio.run(main()) """ _response = await self._raw_client.delete_thread_comments(ids=ids, request_options=request_options) return _response.data async def delete_thread_feedback_scores( self, *, project_name: str, thread_id: str, names: typing.Sequence[str], author: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Delete thread feedback scores Parameters ---------- project_name : str thread_id : str names : typing.Sequence[str] author : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.traces.delete_thread_feedback_scores(project_name='project_name', thread_id='thread_id', names=['names'], ) asyncio.run(main()) """ _response = await self._raw_client.delete_thread_feedback_scores( project_name=project_name, thread_id=thread_id, names=names, author=author, request_options=request_options ) return _response.data async def delete_trace_comments( self, *, ids: typing.Sequence[str], request_options: typing.Optional[RequestOptions] = None ) -> None: """ Delete trace comments Parameters ---------- ids : typing.Sequence[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.traces.delete_trace_comments(ids=['ids'], ) asyncio.run(main()) """ _response = await self._raw_client.delete_trace_comments(ids=ids, request_options=request_options) return _response.data async def delete_trace_feedback_score( self, id: str, *, name: str, author: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Delete trace feedback score Parameters ---------- id : str name : str author : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.traces.delete_trace_feedback_score(id='id', name='name', ) asyncio.run(main()) """ _response = await self._raw_client.delete_trace_feedback_score( id, name=name, author=author, request_options=request_options ) return _response.data async def delete_trace_threads( self, *, thread_ids: typing.Sequence[str], project_name: typing.Optional[str] = OMIT, project_id: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Delete trace threads Parameters ---------- thread_ids : typing.Sequence[str] project_name : typing.Optional[str] If null, project_id must be provided project_id : typing.Optional[str] If null, project_name must be provided request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.traces.delete_trace_threads(thread_ids=['thread_ids'], ) asyncio.run(main()) """ _response = await self._raw_client.delete_trace_threads( thread_ids=thread_ids, project_name=project_name, project_id=project_id, request_options=request_options ) return _response.data async def delete_traces( self, *, ids: typing.Sequence[str], request_options: typing.Optional[RequestOptions] = None ) -> None: """ Delete traces Parameters ---------- ids : typing.Sequence[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.traces.delete_traces(ids=['ids'], ) asyncio.run(main()) """ _response = await self._raw_client.delete_traces(ids=ids, request_options=request_options) return _response.data async def find_feedback_score_names2( self, *, project_id: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None ) -> FeedbackScoreNamesPublic: """ Find Feedback Score names Parameters ---------- project_id : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- FeedbackScoreNamesPublic Feedback Scores resource Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.traces.find_feedback_score_names2() asyncio.run(main()) """ _response = await self._raw_client.find_feedback_score_names2( project_id=project_id, request_options=request_options ) return _response.data async def find_trace_threads_feedback_score_names( self, *, project_id: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None ) -> FeedbackScoreNamesPublic: """ Find Trace Threads Feedback Score names Parameters ---------- project_id : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- FeedbackScoreNamesPublic Find Trace Threads Feedback Score names Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.traces.find_trace_threads_feedback_score_names() asyncio.run(main()) """ _response = await self._raw_client.find_trace_threads_feedback_score_names( project_id=project_id, request_options=request_options ) return _response.data async def get_trace_stats( self, *, project_id: typing.Optional[str] = None, project_name: typing.Optional[str] = None, filters: typing.Optional[str] = None, search: typing.Optional[str] = None, from_time: typing.Optional[dt.datetime] = None, to_time: typing.Optional[dt.datetime] = None, request_options: typing.Optional[RequestOptions] = None, ) -> ProjectStatsPublic: """ Get trace stats Parameters ---------- project_id : typing.Optional[str] project_name : typing.Optional[str] filters : typing.Optional[str] search : typing.Optional[str] from_time : typing.Optional[dt.datetime] to_time : typing.Optional[dt.datetime] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- ProjectStatsPublic Trace stats resource Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.traces.get_trace_stats() asyncio.run(main()) """ _response = await self._raw_client.get_trace_stats( project_id=project_id, project_name=project_name, filters=filters, search=search, from_time=from_time, to_time=to_time, request_options=request_options, ) return _response.data async def get_thread_comment( self, thread_id: str, comment_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> Comment: """ Get thread comment Parameters ---------- thread_id : str comment_id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- Comment Comment resource Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.traces.get_thread_comment(thread_id='threadId', comment_id='commentId', ) asyncio.run(main()) """ _response = await self._raw_client.get_thread_comment(thread_id, comment_id, request_options=request_options) return _response.data async def get_trace_thread_stats( self, *, project_id: typing.Optional[str] = None, project_name: typing.Optional[str] = None, filters: typing.Optional[str] = None, search: typing.Optional[str] = None, from_time: typing.Optional[dt.datetime] = None, to_time: typing.Optional[dt.datetime] = None, request_options: typing.Optional[RequestOptions] = None, ) -> ProjectStatsPublic: """ Get trace thread stats Parameters ---------- project_id : typing.Optional[str] project_name : typing.Optional[str] filters : typing.Optional[str] search : typing.Optional[str] from_time : typing.Optional[dt.datetime] to_time : typing.Optional[dt.datetime] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- ProjectStatsPublic Trace thread stats resource Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.traces.get_trace_thread_stats() asyncio.run(main()) """ _response = await self._raw_client.get_trace_thread_stats( project_id=project_id, project_name=project_name, filters=filters, search=search, from_time=from_time, to_time=to_time, request_options=request_options, ) return _response.data async def get_trace_comment( self, trace_id: str, comment_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> Comment: """ Get trace comment Parameters ---------- trace_id : str comment_id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- Comment Comment resource Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.traces.get_trace_comment(trace_id='traceId', comment_id='commentId', ) asyncio.run(main()) """ _response = await self._raw_client.get_trace_comment(trace_id, comment_id, request_options=request_options) return _response.data async def get_trace_thread( self, *, thread_id: str, project_name: typing.Optional[str] = OMIT, project_id: typing.Optional[str] = OMIT, truncate: typing.Optional[bool] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> TraceThread: """ Get trace thread Parameters ---------- thread_id : str project_name : typing.Optional[str] project_id : typing.Optional[str] truncate : typing.Optional[bool] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- TraceThread Trace thread resource Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.traces.get_trace_thread(thread_id='thread_id', ) asyncio.run(main()) """ _response = await self._raw_client.get_trace_thread( thread_id=thread_id, project_name=project_name, project_id=project_id, truncate=truncate, request_options=request_options, ) return _response.data async def get_trace_threads( self, *, page: typing.Optional[int] = None, size: typing.Optional[int] = None, project_name: typing.Optional[str] = None, project_id: typing.Optional[str] = None, truncate: typing.Optional[bool] = None, strip_attachments: typing.Optional[bool] = None, filters: typing.Optional[str] = None, sorting: typing.Optional[str] = None, search: typing.Optional[str] = None, from_time: typing.Optional[dt.datetime] = None, to_time: typing.Optional[dt.datetime] = None, request_options: typing.Optional[RequestOptions] = None, ) -> TraceThreadPage: """ Get trace threads Parameters ---------- page : typing.Optional[int] size : typing.Optional[int] project_name : typing.Optional[str] project_id : typing.Optional[str] truncate : typing.Optional[bool] strip_attachments : typing.Optional[bool] filters : typing.Optional[str] sorting : typing.Optional[str] search : typing.Optional[str] from_time : typing.Optional[dt.datetime] to_time : typing.Optional[dt.datetime] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- TraceThreadPage Trace threads resource Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.traces.get_trace_threads() asyncio.run(main()) """ _response = await self._raw_client.get_trace_threads( page=page, size=size, project_name=project_name, project_id=project_id, truncate=truncate, strip_attachments=strip_attachments, filters=filters, sorting=sorting, search=search, from_time=from_time, to_time=to_time, request_options=request_options, ) return _response.data async def open_trace_thread( self, *, thread_id: str, project_name: typing.Optional[str] = OMIT, project_id: typing.Optional[str] = OMIT, truncate: typing.Optional[bool] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Open trace thread Parameters ---------- thread_id : str project_name : typing.Optional[str] project_id : typing.Optional[str] truncate : typing.Optional[bool] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.traces.open_trace_thread(thread_id='thread_id', ) asyncio.run(main()) """ _response = await self._raw_client.open_trace_thread( thread_id=thread_id, project_name=project_name, project_id=project_id, truncate=truncate, request_options=request_options, ) return _response.data async def score_batch_of_threads( self, *, scores: typing.Sequence[FeedbackScoreBatchItemThread], request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Batch feedback scoring for threads Parameters ---------- scores : typing.Sequence[FeedbackScoreBatchItemThread] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import AsyncOpikApi from Opik import FeedbackScoreBatchItemThread import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.traces.score_batch_of_threads(scores=[FeedbackScoreBatchItemThread(name='name', value=1.1, source="ui", thread_id='thread_id', )], ) asyncio.run(main()) """ _response = await self._raw_client.score_batch_of_threads(scores=scores, request_options=request_options) return _response.data async def score_batch_of_traces( self, *, scores: typing.Sequence[FeedbackScoreBatchItem], request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Batch feedback scoring for traces Parameters ---------- scores : typing.Sequence[FeedbackScoreBatchItem] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import AsyncOpikApi from Opik import FeedbackScoreBatchItem import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.traces.score_batch_of_traces(scores=[FeedbackScoreBatchItem(name='name', value=1.1, source="ui", id='id', )], ) asyncio.run(main()) """ _response = await self._raw_client.score_batch_of_traces(scores=scores, request_options=request_options) return _response.data async def search_trace_threads( self, *, project_name: typing.Optional[str] = OMIT, project_id: typing.Optional[str] = OMIT, filters: typing.Optional[typing.Sequence[TraceThreadFilter]] = OMIT, last_retrieved_thread_model_id: typing.Optional[str] = OMIT, limit: typing.Optional[int] = OMIT, truncate: typing.Optional[bool] = OMIT, strip_attachments: typing.Optional[bool] = OMIT, from_time: typing.Optional[dt.datetime] = OMIT, to_time: typing.Optional[dt.datetime] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> typing.AsyncIterator[bytes]: """ Search trace threads Parameters ---------- project_name : typing.Optional[str] project_id : typing.Optional[str] filters : typing.Optional[typing.Sequence[TraceThreadFilter]] last_retrieved_thread_model_id : typing.Optional[str] limit : typing.Optional[int] Max number of trace thread to be streamed truncate : typing.Optional[bool] Truncate input, output and metadata to slim payloads strip_attachments : typing.Optional[bool] If true, returns attachment references like [file.png]; if false, downloads and reinjects stripped attachments from_time : typing.Optional[dt.datetime] Filter trace threads created from this time (ISO-8601 format). to_time : typing.Optional[dt.datetime] Filter trace threads created up to this time (ISO-8601 format). If not provided, defaults to current time. Must be after 'from_time'. request_options : typing.Optional[RequestOptions] Request-specific configuration. You can pass in configuration such as `chunk_size`, and more to customize the request and response. Returns ------- typing.AsyncIterator[bytes] Trace threads stream or error during process Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.traces.search_trace_threads() asyncio.run(main()) """ async with self._raw_client.search_trace_threads( project_name=project_name, project_id=project_id, filters=filters, last_retrieved_thread_model_id=last_retrieved_thread_model_id, limit=limit, truncate=truncate, strip_attachments=strip_attachments, from_time=from_time, to_time=to_time, request_options=request_options, ) as r: async for data in r.data: yield data async def search_traces( self, *, project_name: typing.Optional[str] = OMIT, project_id: typing.Optional[str] = OMIT, filters: typing.Optional[typing.Sequence[TraceFilterPublic]] = OMIT, last_retrieved_id: typing.Optional[str] = OMIT, limit: typing.Optional[int] = OMIT, truncate: typing.Optional[bool] = OMIT, strip_attachments: typing.Optional[bool] = OMIT, from_time: typing.Optional[dt.datetime] = OMIT, to_time: typing.Optional[dt.datetime] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> typing.AsyncIterator[bytes]: """ Search traces Parameters ---------- project_name : typing.Optional[str] project_id : typing.Optional[str] filters : typing.Optional[typing.Sequence[TraceFilterPublic]] last_retrieved_id : typing.Optional[str] limit : typing.Optional[int] Max number of traces to be streamed truncate : typing.Optional[bool] Truncate input, output and metadata to slim payloads strip_attachments : typing.Optional[bool] If true, returns attachment references like [file.png]; if false, downloads and reinjects stripped attachments from_time : typing.Optional[dt.datetime] Filter traces created from this time (ISO-8601 format). to_time : typing.Optional[dt.datetime] Filter traces created up to this time (ISO-8601 format). If not provided, defaults to current time. Must be after 'from_time'. request_options : typing.Optional[RequestOptions] Request-specific configuration. You can pass in configuration such as `chunk_size`, and more to customize the request and response. Returns ------- typing.AsyncIterator[bytes] Traces stream or error during process Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.traces.search_traces() asyncio.run(main()) """ async with self._raw_client.search_traces( project_name=project_name, project_id=project_id, filters=filters, last_retrieved_id=last_retrieved_id, limit=limit, truncate=truncate, strip_attachments=strip_attachments, from_time=from_time, to_time=to_time, request_options=request_options, ) as r: async for data in r.data: yield data async def update_thread( self, thread_model_id: str, *, tags: typing.Optional[typing.Sequence[str]] = OMIT, tags_to_add: typing.Optional[typing.Sequence[str]] = OMIT, tags_to_remove: typing.Optional[typing.Sequence[str]] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Update thread Parameters ---------- thread_model_id : str tags : typing.Optional[typing.Sequence[str]] tags_to_add : typing.Optional[typing.Sequence[str]] tags_to_remove : typing.Optional[typing.Sequence[str]] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.traces.update_thread(thread_model_id='threadModelId', ) asyncio.run(main()) """ _response = await self._raw_client.update_thread( thread_model_id, tags=tags, tags_to_add=tags_to_add, tags_to_remove=tags_to_remove, request_options=request_options, ) return _response.data async def update_thread_comment( self, comment_id: str, *, text: str, id: typing.Optional[str] = OMIT, created_at: typing.Optional[dt.datetime] = OMIT, last_updated_at: typing.Optional[dt.datetime] = OMIT, created_by: typing.Optional[str] = OMIT, last_updated_by: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Update thread comment by id Parameters ---------- comment_id : str text : str id : typing.Optional[str] created_at : typing.Optional[dt.datetime] last_updated_at : typing.Optional[dt.datetime] created_by : typing.Optional[str] last_updated_by : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.traces.update_thread_comment(comment_id='commentId', text='text', ) asyncio.run(main()) """ _response = await self._raw_client.update_thread_comment( comment_id, text=text, id=id, created_at=created_at, last_updated_at=last_updated_at, created_by=created_by, last_updated_by=last_updated_by, request_options=request_options, ) return _response.data async def update_trace_comment( self, comment_id: str, *, text: str, id: typing.Optional[str] = OMIT, created_at: typing.Optional[dt.datetime] = OMIT, last_updated_at: typing.Optional[dt.datetime] = OMIT, created_by: typing.Optional[str] = OMIT, last_updated_by: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Update trace comment by id Parameters ---------- comment_id : str text : str id : typing.Optional[str] created_at : typing.Optional[dt.datetime] last_updated_at : typing.Optional[dt.datetime] created_by : typing.Optional[str] last_updated_by : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.traces.update_trace_comment(comment_id='commentId', text='text', ) asyncio.run(main()) """ _response = await self._raw_client.update_trace_comment( comment_id, text=text, id=id, created_at=created_at, last_updated_at=last_updated_at, created_by=created_by, last_updated_by=last_updated_by, request_options=request_options, ) return _response.data