From e35bfdb5f0da9de829f47bf8a69c8027d0b7db89 Mon Sep 17 00:00:00 2001 From: csgulati09 Date: Fri, 7 Jun 2024 16:33:09 +0530 Subject: [PATCH] fix: gitignore to add lib of openai + install require for sdk --- .gitignore | 2 +- portkey_ai/_vendor/openai/lib/.keep | 4 + portkey_ai/_vendor/openai/lib/_old_api.py | 72 ++ portkey_ai/_vendor/openai/lib/_validators.py | 877 +++++++++++++ portkey_ai/_vendor/openai/lib/azure.py | 560 +++++++++ .../_vendor/openai/lib/streaming/__init__.py | 8 + .../openai/lib/streaming/_assistants.py | 1114 +++++++++++++++++ setup.cfg | 7 +- 8 files changed, 2641 insertions(+), 3 deletions(-) create mode 100644 portkey_ai/_vendor/openai/lib/.keep create mode 100644 portkey_ai/_vendor/openai/lib/_old_api.py create mode 100644 portkey_ai/_vendor/openai/lib/_validators.py create mode 100644 portkey_ai/_vendor/openai/lib/azure.py create mode 100644 portkey_ai/_vendor/openai/lib/streaming/__init__.py create mode 100644 portkey_ai/_vendor/openai/lib/streaming/_assistants.py diff --git a/.gitignore b/.gitignore index 754f0220..536df730 100644 --- a/.gitignore +++ b/.gitignore @@ -14,7 +14,7 @@ dist/ downloads/ eggs/ .eggs/ -lib/ +/lib/ lib64/ parts/ sdist/ diff --git a/portkey_ai/_vendor/openai/lib/.keep b/portkey_ai/_vendor/openai/lib/.keep new file mode 100644 index 00000000..5e2c99fd --- /dev/null +++ b/portkey_ai/_vendor/openai/lib/.keep @@ -0,0 +1,4 @@ +File generated from our OpenAPI spec by Stainless. + +This directory can be used to store custom files to expand the SDK. +It is ignored by Stainless code generation and its content (other than this keep file) won't be touched. \ No newline at end of file diff --git a/portkey_ai/_vendor/openai/lib/_old_api.py b/portkey_ai/_vendor/openai/lib/_old_api.py new file mode 100644 index 00000000..929c87e8 --- /dev/null +++ b/portkey_ai/_vendor/openai/lib/_old_api.py @@ -0,0 +1,72 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING, Any +from typing_extensions import override + +from .._utils import LazyProxy +from .._exceptions import OpenAIError + +INSTRUCTIONS = """ + +You tried to access openai.{symbol}, but this is no longer supported in openai>=1.0.0 - see the README at https://github.com/openai/openai-python for the API. + +You can run `openai migrate` to automatically upgrade your codebase to use the 1.0.0 interface. + +Alternatively, you can pin your installation to the old version, e.g. `pip install openai==0.28` + +A detailed migration guide is available here: https://github.com/openai/openai-python/discussions/742 +""" + + +class APIRemovedInV1(OpenAIError): + def __init__(self, *, symbol: str) -> None: + super().__init__(INSTRUCTIONS.format(symbol=symbol)) + + +class APIRemovedInV1Proxy(LazyProxy[Any]): + def __init__(self, *, symbol: str) -> None: + super().__init__() + self._symbol = symbol + + @override + def __load__(self) -> Any: + # return the proxy until it is eventually called so that + # we don't break people that are just checking the attributes + # of a module + return self + + def __call__(self, *_args: Any, **_kwargs: Any) -> Any: + raise APIRemovedInV1(symbol=self._symbol) + + +SYMBOLS = [ + "Edit", + "File", + "Audio", + "Image", + "Model", + "Engine", + "Customer", + "FineTune", + "Embedding", + "Completion", + "Deployment", + "Moderation", + "ErrorObject", + "FineTuningJob", + "ChatCompletion", +] + +# we explicitly tell type checkers that nothing is exported +# from this file so that when we re-export the old symbols +# in `openai/__init__.py` they aren't added to the auto-complete +# suggestions given by editors +if TYPE_CHECKING: + __all__: list[str] = [] +else: + __all__ = SYMBOLS + + +__locals = locals() +for symbol in SYMBOLS: + __locals[symbol] = APIRemovedInV1Proxy(symbol=symbol) diff --git a/portkey_ai/_vendor/openai/lib/_validators.py b/portkey_ai/_vendor/openai/lib/_validators.py new file mode 100644 index 00000000..eb12e4a3 --- /dev/null +++ b/portkey_ai/_vendor/openai/lib/_validators.py @@ -0,0 +1,877 @@ +# pyright: basic +from __future__ import annotations + +import os +import sys +from typing import Any, TypeVar, Callable, Optional, NamedTuple +from typing_extensions import TypeAlias + +from .._extras import pandas as pd + + +class Remediation(NamedTuple): + name: str + immediate_msg: Optional[str] = None + necessary_msg: Optional[str] = None + necessary_fn: Optional[Callable[[Any], Any]] = None + optional_msg: Optional[str] = None + optional_fn: Optional[Callable[[Any], Any]] = None + error_msg: Optional[str] = None + + +OptionalDataFrameT = TypeVar("OptionalDataFrameT", bound="Optional[pd.DataFrame]") + + +def num_examples_validator(df: pd.DataFrame) -> Remediation: + """ + This validator will only print out the number of examples and recommend to the user to increase the number of examples if less than 100. + """ + MIN_EXAMPLES = 100 + optional_suggestion = ( + "" + if len(df) >= MIN_EXAMPLES + else ". In general, we recommend having at least a few hundred examples. We've found that performance tends to linearly increase for every doubling of the number of examples" + ) + immediate_msg = ( + f"\n- Your file contains {len(df)} prompt-completion pairs{optional_suggestion}" + ) + return Remediation(name="num_examples", immediate_msg=immediate_msg) + + +def necessary_column_validator(df: pd.DataFrame, necessary_column: str) -> Remediation: + """ + This validator will ensure that the necessary column is present in the dataframe. + """ + + def lower_case_column(df: pd.DataFrame, column: Any) -> pd.DataFrame: + cols = [c for c in df.columns if str(c).lower() == column] + df.rename(columns={cols[0]: column.lower()}, inplace=True) + return df + + immediate_msg = None + necessary_fn = None + necessary_msg = None + error_msg = None + + if necessary_column not in df.columns: + if necessary_column in [str(c).lower() for c in df.columns]: + + def lower_case_column_creator(df: pd.DataFrame) -> pd.DataFrame: + return lower_case_column(df, necessary_column) + + necessary_fn = lower_case_column_creator + immediate_msg = ( + f"\n- The `{necessary_column}` column/key should be lowercase" + ) + necessary_msg = f"Lower case column name to `{necessary_column}`" + else: + error_msg = f"`{necessary_column}` column/key is missing. Please make sure you name your columns/keys appropriately, then retry" + + return Remediation( + name="necessary_column", + immediate_msg=immediate_msg, + necessary_msg=necessary_msg, + necessary_fn=necessary_fn, + error_msg=error_msg, + ) + + +def additional_column_validator( + df: pd.DataFrame, fields: list[str] = ["prompt", "completion"] +) -> Remediation: + """ + This validator will remove additional columns from the dataframe. + """ + additional_columns = [] + necessary_msg = None + immediate_msg = None + necessary_fn = None # type: ignore + + if len(df.columns) > 2: + additional_columns = [c for c in df.columns if c not in fields] + warn_message = "" + for ac in additional_columns: + dups = [c for c in additional_columns if ac in c] + if len(dups) > 0: + warn_message += f"\n WARNING: Some of the additional columns/keys contain `{ac}` in their name. These will be ignored, and the column/key `{ac}` will be used instead. This could also result from a duplicate column/key in the provided file." + immediate_msg = f"\n- The input file should contain exactly two columns/keys per row. Additional columns/keys present are: {additional_columns}{warn_message}" + necessary_msg = f"Remove additional columns/keys: {additional_columns}" + + def necessary_fn(x: Any) -> Any: + return x[fields] + + return Remediation( + name="additional_column", + immediate_msg=immediate_msg, + necessary_msg=necessary_msg, + necessary_fn=necessary_fn, + ) + + +def non_empty_field_validator( + df: pd.DataFrame, field: str = "completion" +) -> Remediation: + """ + This validator will ensure that no completion is empty. + """ + necessary_msg = None + necessary_fn = None # type: ignore + immediate_msg = None + + if df[field].apply(lambda x: x == "").any() or df[field].isnull().any(): + empty_rows = (df[field] == "") | (df[field].isnull()) + empty_indexes = df.reset_index().index[empty_rows].tolist() + immediate_msg = f"\n- `{field}` column/key should not contain empty strings. These are rows: {empty_indexes}" + + def necessary_fn(x: Any) -> Any: + return x[x[field] != ""].dropna(subset=[field]) + + necessary_msg = f"Remove {len(empty_indexes)} rows with empty {field}s" + + return Remediation( + name=f"empty_{field}", + immediate_msg=immediate_msg, + necessary_msg=necessary_msg, + necessary_fn=necessary_fn, + ) + + +def duplicated_rows_validator( + df: pd.DataFrame, fields: list[str] = ["prompt", "completion"] +) -> Remediation: + """ + This validator will suggest to the user to remove duplicate rows if they exist. + """ + duplicated_rows = df.duplicated(subset=fields) + duplicated_indexes = df.reset_index().index[duplicated_rows].tolist() + immediate_msg = None + optional_msg = None + optional_fn = None # type: ignore + + if len(duplicated_indexes) > 0: + immediate_msg = f"\n- There are {len(duplicated_indexes)} duplicated {'-'.join(fields)} sets. These are rows: {duplicated_indexes}" + optional_msg = f"Remove {len(duplicated_indexes)} duplicate rows" + + def optional_fn(x: Any) -> Any: + return x.drop_duplicates(subset=fields) + + return Remediation( + name="duplicated_rows", + immediate_msg=immediate_msg, + optional_msg=optional_msg, + optional_fn=optional_fn, + ) + + +def long_examples_validator(df: pd.DataFrame) -> Remediation: + """ + This validator will suggest to the user to remove examples that are too long. + """ + immediate_msg = None + optional_msg = None + optional_fn = None # type: ignore + + ft_type = infer_task_type(df) + if ft_type != "open-ended generation": + + def get_long_indexes(d: pd.DataFrame) -> Any: + long_examples = d.apply( + lambda x: len(x.prompt) + len(x.completion) > 10000, axis=1 + ) + return d.reset_index().index[long_examples].tolist() + + long_indexes = get_long_indexes(df) + + if len(long_indexes) > 0: + immediate_msg = f"\n- There are {len(long_indexes)} examples that are very long. These are rows: {long_indexes}\nFor conditional generation, and for classification the examples shouldn't be longer than 2048 tokens." + optional_msg = f"Remove {len(long_indexes)} long examples" + + def optional_fn(x: Any) -> Any: + long_indexes_to_drop = get_long_indexes(x) + if long_indexes != long_indexes_to_drop: + sys.stdout.write( + f"The indices of the long examples has changed as a result of a previously applied recommendation.\nThe {len(long_indexes_to_drop)} long examples to be dropped are now at the following indices: {long_indexes_to_drop}\n" + ) + return x.drop(long_indexes_to_drop) + + return Remediation( + name="long_examples", + immediate_msg=immediate_msg, + optional_msg=optional_msg, + optional_fn=optional_fn, + ) + + +def common_prompt_suffix_validator(df: pd.DataFrame) -> Remediation: + """ + This validator will suggest to add a common suffix to the prompt if one doesn't already exist in case of classification or conditional generation. + """ + error_msg = None + immediate_msg = None + optional_msg = None + optional_fn = None # type: ignore + + # Find a suffix which is not contained within the prompt otherwise + suggested_suffix = "\n\n### =>\n\n" + suffix_options = [ + " ->", + "\n\n###\n\n", + "\n\n===\n\n", + "\n\n---\n\n", + "\n\n===>\n\n", + "\n\n--->\n\n", + ] + for suffix_option in suffix_options: + if suffix_option == " ->": + if df.prompt.str.contains("\n").any(): + continue + if df.prompt.str.contains(suffix_option, regex=False).any(): + continue + suggested_suffix = suffix_option + break + display_suggested_suffix = suggested_suffix.replace("\n", "\\n") + + ft_type = infer_task_type(df) + if ft_type == "open-ended generation": + return Remediation(name="common_suffix") + + def add_suffix(x: Any, suffix: Any) -> Any: + x["prompt"] += suffix + return x + + common_suffix = get_common_xfix(df.prompt, xfix="suffix") + if (df.prompt == common_suffix).all(): + error_msg = f"All prompts are identical: `{common_suffix}`\nConsider leaving the prompts blank if you want to do open-ended generation, otherwise ensure prompts are different" + return Remediation(name="common_suffix", error_msg=error_msg) + + if common_suffix != "": + common_suffix_new_line_handled = common_suffix.replace("\n", "\\n") + immediate_msg = ( + f"\n- All prompts end with suffix `{common_suffix_new_line_handled}`" + ) + if len(common_suffix) > 10: + immediate_msg += f". This suffix seems very long. Consider replacing with a shorter suffix, such as `{display_suggested_suffix}`" + if ( + df.prompt.str[: -len(common_suffix)] + .str.contains(common_suffix, regex=False) + .any() + ): + immediate_msg += f"\n WARNING: Some of your prompts contain the suffix `{common_suffix}` more than once. We strongly suggest that you review your prompts and add a unique suffix" + + else: + immediate_msg = "\n- Your data does not contain a common separator at the end of your prompts. Having a separator string appended to the end of the prompt makes it clearer to the fine-tuned model where the completion should begin. See https://platform.openai.com/docs/guides/fine-tuning/preparing-your-dataset for more detail and examples. If you intend to do open-ended generation, then you should leave the prompts empty" + + if common_suffix == "": + optional_msg = ( + f"Add a suffix separator `{display_suggested_suffix}` to all prompts" + ) + + def optional_fn(x: Any) -> Any: + return add_suffix(x, suggested_suffix) + + return Remediation( + name="common_completion_suffix", + immediate_msg=immediate_msg, + optional_msg=optional_msg, + optional_fn=optional_fn, + error_msg=error_msg, + ) + + +def common_prompt_prefix_validator(df: pd.DataFrame) -> Remediation: + """ + This validator will suggest to remove a common prefix from the prompt if a long one exist. + """ + MAX_PREFIX_LEN = 12 + + immediate_msg = None + optional_msg = None + optional_fn = None # type: ignore + + common_prefix = get_common_xfix(df.prompt, xfix="prefix") + if common_prefix == "": + return Remediation(name="common_prefix") + + def remove_common_prefix(x: Any, prefix: Any) -> Any: + x["prompt"] = x["prompt"].str[len(prefix) :] + return x + + if (df.prompt == common_prefix).all(): + # already handled by common_suffix_validator + return Remediation(name="common_prefix") + + if common_prefix != "": + immediate_msg = f"\n- All prompts start with prefix `{common_prefix}`" + if MAX_PREFIX_LEN < len(common_prefix): + immediate_msg += ". Fine-tuning doesn't require the instruction specifying the task, or a few-shot example scenario. Most of the time you should only add the input data into the prompt, and the desired output into the completion" + optional_msg = f"Remove prefix `{common_prefix}` from all prompts" + + def optional_fn(x: Any) -> Any: + return remove_common_prefix(x, common_prefix) + + return Remediation( + name="common_prompt_prefix", + immediate_msg=immediate_msg, + optional_msg=optional_msg, + optional_fn=optional_fn, + ) + + +def common_completion_prefix_validator(df: pd.DataFrame) -> Remediation: + """ + This validator will suggest to remove a common prefix from the completion if a long one exist. + """ + MAX_PREFIX_LEN = 5 + + common_prefix = get_common_xfix(df.completion, xfix="prefix") + ws_prefix = len(common_prefix) > 0 and common_prefix[0] == " " + if len(common_prefix) < MAX_PREFIX_LEN: + return Remediation(name="common_prefix") + + def remove_common_prefix(x: Any, prefix: Any, ws_prefix: Any) -> Any: + x["completion"] = x["completion"].str[len(prefix) :] + if ws_prefix: + # keep the single whitespace as prefix + x["completion"] = f" {x['completion']}" + return x + + if (df.completion == common_prefix).all(): + # already handled by common_suffix_validator + return Remediation(name="common_prefix") + + immediate_msg = f"\n- All completions start with prefix `{common_prefix}`. Most of the time you should only add the output data into the completion, without any prefix" + optional_msg = f"Remove prefix `{common_prefix}` from all completions" + + def optional_fn(x: Any) -> Any: + return remove_common_prefix(x, common_prefix, ws_prefix) + + return Remediation( + name="common_completion_prefix", + immediate_msg=immediate_msg, + optional_msg=optional_msg, + optional_fn=optional_fn, + ) + + +def common_completion_suffix_validator(df: pd.DataFrame) -> Remediation: + """ + This validator will suggest to add a common suffix to the completion if one doesn't already exist in case of classification or conditional generation. + """ + error_msg = None + immediate_msg = None + optional_msg = None + optional_fn = None # type: ignore + + ft_type = infer_task_type(df) + if ft_type == "open-ended generation" or ft_type == "classification": + return Remediation(name="common_suffix") + + common_suffix = get_common_xfix(df.completion, xfix="suffix") + if (df.completion == common_suffix).all(): + error_msg = f"All completions are identical: `{common_suffix}`\nEnsure completions are different, otherwise the model will just repeat `{common_suffix}`" + return Remediation(name="common_suffix", error_msg=error_msg) + + # Find a suffix which is not contained within the completion otherwise + suggested_suffix = " [END]" + suffix_options = [ + "\n", + ".", + " END", + "***", + "+++", + "&&&", + "$$$", + "@@@", + "%%%", + ] + for suffix_option in suffix_options: + if df.completion.str.contains(suffix_option, regex=False).any(): + continue + suggested_suffix = suffix_option + break + display_suggested_suffix = suggested_suffix.replace("\n", "\\n") + + def add_suffix(x: Any, suffix: Any) -> Any: + x["completion"] += suffix + return x + + if common_suffix != "": + common_suffix_new_line_handled = common_suffix.replace("\n", "\\n") + immediate_msg = ( + f"\n- All completions end with suffix `{common_suffix_new_line_handled}`" + ) + if len(common_suffix) > 10: + immediate_msg += f". This suffix seems very long. Consider replacing with a shorter suffix, such as `{display_suggested_suffix}`" + if ( + df.completion.str[: -len(common_suffix)] + .str.contains(common_suffix, regex=False) + .any() + ): + immediate_msg += f"\n WARNING: Some of your completions contain the suffix `{common_suffix}` more than once. We suggest that you review your completions and add a unique ending" + + else: + immediate_msg = "\n- Your data does not contain a common ending at the end of your completions. Having a common ending string appended to the end of the completion makes it clearer to the fine-tuned model where the completion should end. See https://platform.openai.com/docs/guides/fine-tuning/preparing-your-dataset for more detail and examples." + + if common_suffix == "": + optional_msg = ( + f"Add a suffix ending `{display_suggested_suffix}` to all completions" + ) + + def optional_fn(x: Any) -> Any: + return add_suffix(x, suggested_suffix) + + return Remediation( + name="common_completion_suffix", + immediate_msg=immediate_msg, + optional_msg=optional_msg, + optional_fn=optional_fn, + error_msg=error_msg, + ) + + +def completions_space_start_validator(df: pd.DataFrame) -> Remediation: + """ + This validator will suggest to add a space at the start of the completion if it doesn't already exist. This helps with tokenization. + """ + + def add_space_start(x: Any) -> Any: + x["completion"] = x["completion"].apply( + lambda s: ("" if s.startswith(" ") else " ") + s + ) + return x + + optional_msg = None + optional_fn = None + immediate_msg = None + + if df.completion.str[:1].nunique() != 1 or df.completion.values[0][0] != " ": + immediate_msg = "\n- The completion should start with a whitespace character (` `). This tends to produce better results due to the tokenization we use. See https://platform.openai.com/docs/guides/fine-tuning/preparing-your-dataset for more details" + optional_msg = "Add a whitespace character to the beginning of the completion" + optional_fn = add_space_start + return Remediation( + name="completion_space_start", + immediate_msg=immediate_msg, + optional_msg=optional_msg, + optional_fn=optional_fn, + ) + + +def lower_case_validator(df: pd.DataFrame, column: Any) -> Remediation | None: + """ + This validator will suggest to lowercase the column values, if more than a third of letters are uppercase. + """ + + def lower_case(x: Any) -> Any: + x[column] = x[column].str.lower() + return x + + count_upper = ( + df[column] + .apply(lambda x: sum(1 for c in x if c.isalpha() and c.isupper())) + .sum() + ) + count_lower = ( + df[column] + .apply(lambda x: sum(1 for c in x if c.isalpha() and c.islower())) + .sum() + ) + + if count_upper * 2 > count_lower: + return Remediation( + name="lower_case", + immediate_msg=f"\n- More than a third of your `{column}` column/key is uppercase. Uppercase {column}s tends to perform worse than a mixture of case encountered in normal language. We recommend to lower case the data if that makes sense in your domain. See https://platform.openai.com/docs/guides/fine-tuning/preparing-your-dataset for more details", + optional_msg=f"Lowercase all your data in column/key `{column}`", + optional_fn=lower_case, + ) + return None + + +def read_any_format( + fname: str, fields: list[str] = ["prompt", "completion"] +) -> tuple[pd.DataFrame | None, Remediation]: + """ + This function will read a file saved in .csv, .json, .txt, .xlsx or .tsv format using pandas. + - for .xlsx it will read the first sheet + - for .txt it will assume completions and split on newline + """ + remediation = None + necessary_msg = None + immediate_msg = None + error_msg = None + df = None + + if os.path.isfile(fname): + try: + if fname.lower().endswith(".csv") or fname.lower().endswith(".tsv"): + file_extension_str, separator = ( + ("CSV", ",") if fname.lower().endswith(".csv") else ("TSV", "\t") + ) + immediate_msg = f"\n- Based on your file extension, your file is formatted as a {file_extension_str} file" + necessary_msg = ( + f"Your format `{file_extension_str}` will be converted to `JSONL`" + ) + df = pd.read_csv(fname, sep=separator, dtype=str).fillna("") + elif fname.lower().endswith(".xlsx"): + immediate_msg = "\n- Based on your file extension, your file is formatted as an Excel file" + necessary_msg = "Your format `XLSX` will be converted to `JSONL`" + xls = pd.ExcelFile(fname) + sheets = xls.sheet_names + if len(sheets) > 1: + immediate_msg += "\n- Your Excel file contains more than one sheet. Please either save as csv or ensure all data is present in the first sheet. WARNING: Reading only the first sheet..." + df = pd.read_excel(fname, dtype=str).fillna("") + elif fname.lower().endswith(".txt"): + immediate_msg = ( + "\n- Based on your file extension, you provided a text file" + ) + necessary_msg = "Your format `TXT` will be converted to `JSONL`" + with open(fname, "r") as f: + content = f.read() + df = pd.DataFrame( + [["", line] for line in content.split("\n")], + columns=fields, + dtype=str, + ).fillna("") + elif fname.lower().endswith(".jsonl"): + df = pd.read_json(fname, lines=True, dtype=str).fillna("") # type: ignore + if len(df) == 1: # type: ignore + # this is NOT what we expect for a .jsonl file + immediate_msg = "\n- Your JSONL file appears to be in a JSON format. Your file will be converted to JSONL format" + necessary_msg = "Your format `JSON` will be converted to `JSONL`" + df = pd.read_json(fname, dtype=str).fillna("") # type: ignore + else: + pass # this is what we expect for a .jsonl file + elif fname.lower().endswith(".json"): + try: + # to handle case where .json file is actually a .jsonl file + df = pd.read_json(fname, lines=True, dtype=str).fillna("") # type: ignore + if len(df) == 1: # type: ignore + # this code path corresponds to a .json file that has one line + df = pd.read_json(fname, dtype=str).fillna("") # type: ignore + else: + # this is NOT what we expect for a .json file + immediate_msg = "\n- Your JSON file appears to be in a JSONL format. Your file will be converted to JSONL format" + necessary_msg = ( + "Your format `JSON` will be converted to `JSONL`" + ) + except ValueError: + # this code path corresponds to a .json file that has multiple lines (i.e. it is indented) + df = pd.read_json(fname, dtype=str).fillna("") # type: ignore + else: + error_msg = "Your file must have one of the following extensions: .CSV, .TSV, .XLSX, .TXT, .JSON or .JSONL" + if "." in fname: + error_msg += f" Your file `{fname}` ends with the extension `.{fname.split('.')[-1]}` which is not supported." + else: + error_msg += f" Your file `{fname}` is missing a file extension." + + except (ValueError, TypeError): + file_extension_str = fname.split(".")[-1].upper() + error_msg = f"Your file `{fname}` does not appear to be in valid {file_extension_str} format. Please ensure your file is formatted as a valid {file_extension_str} file." + + else: + error_msg = f"File {fname} does not exist." + + remediation = Remediation( + name="read_any_format", + necessary_msg=necessary_msg, + immediate_msg=immediate_msg, + error_msg=error_msg, + ) + return df, remediation + + +def format_inferrer_validator(df: pd.DataFrame) -> Remediation: + """ + This validator will infer the likely fine-tuning format of the data, and display it to the user if it is classification. + It will also suggest to use ada and explain train/validation split benefits. + """ + ft_type = infer_task_type(df) + immediate_msg = None + if ft_type == "classification": + immediate_msg = f"\n- Based on your data it seems like you're trying to fine-tune a model for {ft_type}\n- For classification, we recommend you try one of the faster and cheaper models, such as `ada`\n- For classification, you can estimate the expected model performance by keeping a held out dataset, which is not used for training" + return Remediation(name="num_examples", immediate_msg=immediate_msg) + + +def apply_necessary_remediation( + df: OptionalDataFrameT, remediation: Remediation +) -> OptionalDataFrameT: + """ + This function will apply a necessary remediation to a dataframe, or print an error message if one exists. + """ + if remediation.error_msg is not None: + sys.stderr.write( + f"\n\nERROR in {remediation.name} validator: {remediation.error_msg}\n\nAborting..." + ) + sys.exit(1) + if remediation.immediate_msg is not None: + sys.stdout.write(remediation.immediate_msg) + if remediation.necessary_fn is not None: + df = remediation.necessary_fn(df) + return df + + +def accept_suggestion(input_text: str, auto_accept: bool) -> bool: + sys.stdout.write(input_text) + if auto_accept: + sys.stdout.write("Y\n") + return True + return input().lower() != "n" + + +def apply_optional_remediation( + df: pd.DataFrame, remediation: Remediation, auto_accept: bool +) -> tuple[pd.DataFrame, bool]: + """ + This function will apply an optional remediation to a dataframe, based on the user input. + """ + optional_applied = False + input_text = f"- [Recommended] {remediation.optional_msg} [Y/n]: " + if remediation.optional_msg is not None: + if accept_suggestion(input_text, auto_accept): + assert remediation.optional_fn is not None + df = remediation.optional_fn(df) + optional_applied = True + if remediation.necessary_msg is not None: + sys.stdout.write(f"- [Necessary] {remediation.necessary_msg}\n") + return df, optional_applied + + +def estimate_fine_tuning_time(df: pd.DataFrame) -> None: + """ + Estimate the time it'll take to fine-tune the dataset + """ + ft_format = infer_task_type(df) + expected_time = 1.0 + if ft_format == "classification": + num_examples = len(df) + expected_time = num_examples * 1.44 + else: + size = df.memory_usage(index=True).sum() + expected_time = size * 0.0515 + + def format_time(time: float) -> str: + if time < 60: + return f"{round(time, 2)} seconds" + elif time < 3600: + return f"{round(time / 60, 2)} minutes" + elif time < 86400: + return f"{round(time / 3600, 2)} hours" + else: + return f"{round(time / 86400, 2)} days" + + time_string = format_time(expected_time + 140) + sys.stdout.write( + f"Once your model starts training, it'll approximately take {time_string} to train a `curie` model, and less for `ada` and `babbage`. Queue will approximately take half an hour per job ahead of you.\n" + ) + + +def get_outfnames(fname: str, split: bool) -> list[str]: + suffixes = ["_train", "_valid"] if split else [""] + i = 0 + while True: + index_suffix = f" ({i})" if i > 0 else "" + candidate_fnames = [ + f"{os.path.splitext(fname)[0]}_prepared{suffix}{index_suffix}.jsonl" + for suffix in suffixes + ] + if not any(os.path.isfile(f) for f in candidate_fnames): + return candidate_fnames + i += 1 + + +def get_classification_hyperparams(df: pd.DataFrame) -> tuple[int, object]: + n_classes = df.completion.nunique() + pos_class = None + if n_classes == 2: + pos_class = df.completion.value_counts().index[0] + return n_classes, pos_class + + +def write_out_file( + df: pd.DataFrame, fname: str, any_remediations: bool, auto_accept: bool +) -> None: + """ + This function will write out a dataframe to a file, if the user would like to proceed, and also offer a fine-tuning command with the newly created file. + For classification it will optionally ask the user if they would like to split the data into train/valid files, and modify the suggested command to include the valid set. + """ + ft_format = infer_task_type(df) + common_prompt_suffix = get_common_xfix(df.prompt, xfix="suffix") + common_completion_suffix = get_common_xfix(df.completion, xfix="suffix") + + split = False + input_text = "- [Recommended] Would you like to split into training and validation set? [Y/n]: " + if ft_format == "classification": + if accept_suggestion(input_text, auto_accept): + split = True + + additional_params = "" + common_prompt_suffix_new_line_handled = common_prompt_suffix.replace("\n", "\\n") + common_completion_suffix_new_line_handled = common_completion_suffix.replace( + "\n", "\\n" + ) + optional_ending_string = ( + f' Make sure to include `stop=["{common_completion_suffix_new_line_handled}"]` so that the generated texts ends at the expected place.' + if len(common_completion_suffix_new_line_handled) > 0 + else "" + ) + + input_text = "\n\nYour data will be written to a new JSONL file. Proceed [Y/n]: " + + if not any_remediations and not split: + sys.stdout.write( + f'\nYou can use your file for fine-tuning:\n> openai api fine_tunes.create -t "{fname}"{additional_params}\n\nAfter you’ve fine-tuned a model, remember that your prompt has to end with the indicator string `{common_prompt_suffix_new_line_handled}` for the model to start generating completions, rather than continuing with the prompt.{optional_ending_string}\n' + ) + estimate_fine_tuning_time(df) + + elif accept_suggestion(input_text, auto_accept): + fnames = get_outfnames(fname, split) + if split: + assert len(fnames) == 2 and "train" in fnames[0] and "valid" in fnames[1] + MAX_VALID_EXAMPLES = 1000 + n_train = max(len(df) - MAX_VALID_EXAMPLES, int(len(df) * 0.8)) + df_train = df.sample(n=n_train, random_state=42) + df_valid = df.drop(df_train.index) + df_train[["prompt", "completion"]].to_json( # type: ignore + fnames[0], lines=True, orient="records", force_ascii=False, indent=None + ) + df_valid[["prompt", "completion"]].to_json( + fnames[1], lines=True, orient="records", force_ascii=False, indent=None + ) + + n_classes, pos_class = get_classification_hyperparams(df) + additional_params += " --compute_classification_metrics" + if n_classes == 2: + additional_params += f' --classification_positive_class "{pos_class}"' + else: + additional_params += f" --classification_n_classes {n_classes}" + else: + assert len(fnames) == 1 + df[["prompt", "completion"]].to_json( + fnames[0], lines=True, orient="records", force_ascii=False, indent=None + ) + + # Add -v VALID_FILE if we split the file into train / valid + files_string = ("s" if split else "") + " to `" + ("` and `".join(fnames)) + valid_string = f' -v "{fnames[1]}"' if split else "" + separator_reminder = ( + "" + if len(common_prompt_suffix_new_line_handled) == 0 + else f"After you’ve fine-tuned a model, remember that your prompt has to end with the indicator string `{common_prompt_suffix_new_line_handled}` for the model to start generating completions, rather than continuing with the prompt." + ) + sys.stdout.write( + f'\nWrote modified file{files_string}`\nFeel free to take a look!\n\nNow use that file when fine-tuning:\n> openai api fine_tunes.create -t "{fnames[0]}"{valid_string}{additional_params}\n\n{separator_reminder}{optional_ending_string}\n' + ) + estimate_fine_tuning_time(df) + else: + sys.stdout.write("Aborting... did not write the file\n") + + +def infer_task_type(df: pd.DataFrame) -> str: + """ + Infer the likely fine-tuning task type from the data + """ + CLASSIFICATION_THRESHOLD = 3 # min_average instances of each class + if sum(df.prompt.str.len()) == 0: + return "open-ended generation" + + if len(df.completion.unique()) < len(df) / CLASSIFICATION_THRESHOLD: + return "classification" + + return "conditional generation" + + +def get_common_xfix(series: Any, xfix: str = "suffix") -> str: + """ + Finds the longest common suffix or prefix of all the values in a series + """ + common_xfix = "" + while True: + common_xfixes = ( + series.str[-(len(common_xfix) + 1) :] + if xfix == "suffix" + else series.str[: len(common_xfix) + 1] + ) # first few or last few characters + if ( + common_xfixes.nunique() != 1 + ): # we found the character at which we don't have a unique xfix anymore + break + elif ( + common_xfix == common_xfixes.values[0] + ): # the entire first row is a prefix of every other row + break + else: # the first or last few characters are still common across all rows - let's try to add one more + common_xfix = common_xfixes.values[0] + return common_xfix + + +Validator: TypeAlias = "Callable[[pd.DataFrame], Remediation | None]" + + +def get_validators() -> list[Validator]: + return [ + num_examples_validator, + lambda x: necessary_column_validator(x, "prompt"), + lambda x: necessary_column_validator(x, "completion"), + additional_column_validator, + non_empty_field_validator, + format_inferrer_validator, + duplicated_rows_validator, + long_examples_validator, + lambda x: lower_case_validator(x, "prompt"), + lambda x: lower_case_validator(x, "completion"), + common_prompt_suffix_validator, + common_prompt_prefix_validator, + common_completion_prefix_validator, + common_completion_suffix_validator, + completions_space_start_validator, + ] + + +def apply_validators( + df: pd.DataFrame, + fname: str, + remediation: Remediation | None, + validators: list[Validator], + auto_accept: bool, + write_out_file_func: Callable[..., Any], +) -> None: + optional_remediations: list[Remediation] = [] + if remediation is not None: + optional_remediations.append(remediation) + for validator in validators: + remediation = validator(df) + if remediation is not None: + optional_remediations.append(remediation) + df = apply_necessary_remediation(df, remediation) + + any_optional_or_necessary_remediations = any( + [ + remediation + for remediation in optional_remediations + if remediation.optional_msg is not None + or remediation.necessary_msg is not None + ] + ) + any_necessary_applied = any( + [ + remediation + for remediation in optional_remediations + if remediation.necessary_msg is not None + ] + ) + any_optional_applied = False + + if any_optional_or_necessary_remediations: + sys.stdout.write( + "\n\nBased on the analysis we will perform the following actions:\n" + ) + for remediation in optional_remediations: + df, optional_applied = apply_optional_remediation( + df, remediation, auto_accept + ) + any_optional_applied = any_optional_applied or optional_applied + else: + sys.stdout.write("\n\nNo remediations found.\n") + + any_optional_or_necessary_applied = any_optional_applied or any_necessary_applied + + write_out_file_func(df, fname, any_optional_or_necessary_applied, auto_accept) diff --git a/portkey_ai/_vendor/openai/lib/azure.py b/portkey_ai/_vendor/openai/lib/azure.py new file mode 100644 index 00000000..aab87394 --- /dev/null +++ b/portkey_ai/_vendor/openai/lib/azure.py @@ -0,0 +1,560 @@ +from __future__ import annotations + +import os +import inspect +from typing import Any, Union, Mapping, TypeVar, Callable, Awaitable, overload +from typing_extensions import Self, override + +import httpx + +from .._types import NOT_GIVEN, Omit, Timeout, NotGiven +from .._utils import is_given, is_mapping +from .._client import OpenAI, AsyncOpenAI +from .._models import FinalRequestOptions +from .._streaming import Stream, AsyncStream +from .._exceptions import OpenAIError +from .._base_client import DEFAULT_MAX_RETRIES, BaseClient + +_deployments_endpoints = set( + [ + "/completions", + "/chat/completions", + "/embeddings", + "/audio/transcriptions", + "/audio/translations", + "/audio/speech", + "/images/generations", + ] +) + + +AzureADTokenProvider = Callable[[], str] +AsyncAzureADTokenProvider = Callable[[], "str | Awaitable[str]"] +_HttpxClientT = TypeVar("_HttpxClientT", bound=Union[httpx.Client, httpx.AsyncClient]) +_DefaultStreamT = TypeVar("_DefaultStreamT", bound=Union[Stream[Any], AsyncStream[Any]]) + + +# we need to use a sentinel API key value for Azure AD +# as we don't want to make the `api_key` in the main client Optional +# and Azure AD tokens may be retrieved on a per-request basis +API_KEY_SENTINEL = "".join(["<", "missing API key", ">"]) + + +class MutuallyExclusiveAuthError(OpenAIError): + def __init__(self) -> None: + super().__init__( + "The `api_key`, `azure_ad_token` and `azure_ad_token_provider` arguments are mutually exclusive; Only one can be passed at a time" + ) + + +class BaseAzureClient(BaseClient[_HttpxClientT, _DefaultStreamT]): + @override + def _build_request( + self, + options: FinalRequestOptions, + ) -> httpx.Request: + if options.url in _deployments_endpoints and is_mapping(options.json_data): + model = options.json_data.get("model") + if model is not None and not "/deployments" in str(self.base_url): + options.url = f"/deployments/{model}{options.url}" + + return super()._build_request(options) + + +class AzureOpenAI(BaseAzureClient[httpx.Client, Stream[Any]], OpenAI): + @overload + def __init__( + self, + *, + azure_endpoint: str, + azure_deployment: str | None = None, + api_version: str | None = None, + api_key: str | None = None, + azure_ad_token: str | None = None, + azure_ad_token_provider: AzureADTokenProvider | None = None, + organization: str | None = None, + timeout: float | Timeout | None | NotGiven = NOT_GIVEN, + max_retries: int = DEFAULT_MAX_RETRIES, + default_headers: Mapping[str, str] | None = None, + default_query: Mapping[str, object] | None = None, + http_client: httpx.Client | None = None, + _strict_response_validation: bool = False, + ) -> None: + ... + + @overload + def __init__( + self, + *, + azure_deployment: str | None = None, + api_version: str | None = None, + api_key: str | None = None, + azure_ad_token: str | None = None, + azure_ad_token_provider: AzureADTokenProvider | None = None, + organization: str | None = None, + timeout: float | Timeout | None | NotGiven = NOT_GIVEN, + max_retries: int = DEFAULT_MAX_RETRIES, + default_headers: Mapping[str, str] | None = None, + default_query: Mapping[str, object] | None = None, + http_client: httpx.Client | None = None, + _strict_response_validation: bool = False, + ) -> None: + ... + + @overload + def __init__( + self, + *, + base_url: str, + api_version: str | None = None, + api_key: str | None = None, + azure_ad_token: str | None = None, + azure_ad_token_provider: AzureADTokenProvider | None = None, + organization: str | None = None, + timeout: float | Timeout | None | NotGiven = NOT_GIVEN, + max_retries: int = DEFAULT_MAX_RETRIES, + default_headers: Mapping[str, str] | None = None, + default_query: Mapping[str, object] | None = None, + http_client: httpx.Client | None = None, + _strict_response_validation: bool = False, + ) -> None: + ... + + def __init__( + self, + *, + api_version: str | None = None, + azure_endpoint: str | None = None, + azure_deployment: str | None = None, + api_key: str | None = None, + azure_ad_token: str | None = None, + azure_ad_token_provider: AzureADTokenProvider | None = None, + organization: str | None = None, + project: str | None = None, + base_url: str | None = None, + timeout: float | Timeout | None | NotGiven = NOT_GIVEN, + max_retries: int = DEFAULT_MAX_RETRIES, + default_headers: Mapping[str, str] | None = None, + default_query: Mapping[str, object] | None = None, + http_client: httpx.Client | None = None, + _strict_response_validation: bool = False, + ) -> None: + """Construct a new synchronous azure openai client instance. + + This automatically infers the following arguments from their corresponding environment variables if they are not provided: + - `api_key` from `AZURE_OPENAI_API_KEY` + - `organization` from `OPENAI_ORG_ID` + - `project` from `OPENAI_PROJECT_ID` + - `azure_ad_token` from `AZURE_OPENAI_AD_TOKEN` + - `api_version` from `OPENAI_API_VERSION` + - `azure_endpoint` from `AZURE_OPENAI_ENDPOINT` + + Args: + azure_endpoint: Your Azure endpoint, including the resource, e.g. `https://example-resource.azure.openai.com/` + + azure_ad_token: Your Azure Active Directory token, https://www.microsoft.com/en-us/security/business/identity-access/microsoft-entra-id + + azure_ad_token_provider: A function that returns an Azure Active Directory token, will be invoked on every request. + + azure_deployment: A model deployment, if given sets the base client URL to include `/deployments/{azure_deployment}`. + Note: this means you won't be able to use non-deployment endpoints. Not supported with Assistants APIs. + """ + if api_key is None: + api_key = os.environ.get("AZURE_OPENAI_API_KEY") + + if azure_ad_token is None: + azure_ad_token = os.environ.get("AZURE_OPENAI_AD_TOKEN") + + if ( + api_key is None + and azure_ad_token is None + and azure_ad_token_provider is None + ): + raise OpenAIError( + "Missing credentials. Please pass one of `api_key`, `azure_ad_token`, `azure_ad_token_provider`, or the `AZURE_OPENAI_API_KEY` or `AZURE_OPENAI_AD_TOKEN` environment variables." + ) + + if api_version is None: + api_version = os.environ.get("OPENAI_API_VERSION") + + if api_version is None: + raise ValueError( + "Must provide either the `api_version` argument or the `OPENAI_API_VERSION` environment variable" + ) + + if default_query is None: + default_query = {"api-version": api_version} + else: + default_query = {**default_query, "api-version": api_version} + + if base_url is None: + if azure_endpoint is None: + azure_endpoint = os.environ.get("AZURE_OPENAI_ENDPOINT") + + if azure_endpoint is None: + raise ValueError( + "Must provide one of the `base_url` or `azure_endpoint` arguments, or the `AZURE_OPENAI_ENDPOINT` environment variable" + ) + + if azure_deployment is not None: + base_url = f"{azure_endpoint}/openai/deployments/{azure_deployment}" + else: + base_url = f"{azure_endpoint}/openai" + else: + if azure_endpoint is not None: + raise ValueError("base_url and azure_endpoint are mutually exclusive") + + if api_key is None: + # define a sentinel value to avoid any typing issues + api_key = API_KEY_SENTINEL + + super().__init__( + api_key=api_key, + organization=organization, + project=project, + base_url=base_url, + timeout=timeout, + max_retries=max_retries, + default_headers=default_headers, + default_query=default_query, + http_client=http_client, + _strict_response_validation=_strict_response_validation, + ) + self._api_version = api_version + self._azure_ad_token = azure_ad_token + self._azure_ad_token_provider = azure_ad_token_provider + + @override + def copy( + self, + *, + api_key: str | None = None, + organization: str | None = None, + project: str | None = None, + api_version: str | None = None, + azure_ad_token: str | None = None, + azure_ad_token_provider: AzureADTokenProvider | None = None, + base_url: str | httpx.URL | None = None, + timeout: float | Timeout | None | NotGiven = NOT_GIVEN, + http_client: httpx.Client | None = None, + max_retries: int | NotGiven = NOT_GIVEN, + default_headers: Mapping[str, str] | None = None, + set_default_headers: Mapping[str, str] | None = None, + default_query: Mapping[str, object] | None = None, + set_default_query: Mapping[str, object] | None = None, + _extra_kwargs: Mapping[str, Any] = {}, + ) -> Self: + """ + Create a new client instance re-using the same options given to the current client with optional overriding. + """ + return super().copy( + api_key=api_key, + organization=organization, + project=project, + base_url=base_url, + timeout=timeout, + http_client=http_client, + max_retries=max_retries, + default_headers=default_headers, + set_default_headers=set_default_headers, + default_query=default_query, + set_default_query=set_default_query, + _extra_kwargs={ + "api_version": api_version or self._api_version, + "azure_ad_token": azure_ad_token or self._azure_ad_token, + "azure_ad_token_provider": azure_ad_token_provider + or self._azure_ad_token_provider, + **_extra_kwargs, + }, + ) + + with_options = copy + + def _get_azure_ad_token(self) -> str | None: + if self._azure_ad_token is not None: + return self._azure_ad_token + + provider = self._azure_ad_token_provider + if provider is not None: + token = provider() + if not token or not isinstance( + token, str + ): # pyright: ignore[reportUnnecessaryIsInstance] + raise ValueError( + f"Expected `azure_ad_token_provider` argument to return a string but it returned {token}", + ) + return token + + return None + + @override + def _prepare_options(self, options: FinalRequestOptions) -> None: + headers: dict[str, str | Omit] = ( + {**options.headers} if is_given(options.headers) else {} + ) + options.headers = headers + + azure_ad_token = self._get_azure_ad_token() + if azure_ad_token is not None: + if headers.get("Authorization") is None: + headers["Authorization"] = f"Bearer {azure_ad_token}" + elif self.api_key is not API_KEY_SENTINEL: + if headers.get("api-key") is None: + headers["api-key"] = self.api_key + else: + # should never be hit + raise ValueError("Unable to handle auth") + + return super()._prepare_options(options) + + +class AsyncAzureOpenAI( + BaseAzureClient[httpx.AsyncClient, AsyncStream[Any]], AsyncOpenAI +): + @overload + def __init__( + self, + *, + azure_endpoint: str, + azure_deployment: str | None = None, + api_version: str | None = None, + api_key: str | None = None, + azure_ad_token: str | None = None, + azure_ad_token_provider: AsyncAzureADTokenProvider | None = None, + organization: str | None = None, + project: str | None = None, + timeout: float | Timeout | None | NotGiven = NOT_GIVEN, + max_retries: int = DEFAULT_MAX_RETRIES, + default_headers: Mapping[str, str] | None = None, + default_query: Mapping[str, object] | None = None, + http_client: httpx.AsyncClient | None = None, + _strict_response_validation: bool = False, + ) -> None: + ... + + @overload + def __init__( + self, + *, + azure_deployment: str | None = None, + api_version: str | None = None, + api_key: str | None = None, + azure_ad_token: str | None = None, + azure_ad_token_provider: AsyncAzureADTokenProvider | None = None, + organization: str | None = None, + project: str | None = None, + timeout: float | Timeout | None | NotGiven = NOT_GIVEN, + max_retries: int = DEFAULT_MAX_RETRIES, + default_headers: Mapping[str, str] | None = None, + default_query: Mapping[str, object] | None = None, + http_client: httpx.AsyncClient | None = None, + _strict_response_validation: bool = False, + ) -> None: + ... + + @overload + def __init__( + self, + *, + base_url: str, + api_version: str | None = None, + api_key: str | None = None, + azure_ad_token: str | None = None, + azure_ad_token_provider: AsyncAzureADTokenProvider | None = None, + organization: str | None = None, + project: str | None = None, + timeout: float | Timeout | None | NotGiven = NOT_GIVEN, + max_retries: int = DEFAULT_MAX_RETRIES, + default_headers: Mapping[str, str] | None = None, + default_query: Mapping[str, object] | None = None, + http_client: httpx.AsyncClient | None = None, + _strict_response_validation: bool = False, + ) -> None: + ... + + def __init__( + self, + *, + azure_endpoint: str | None = None, + azure_deployment: str | None = None, + api_version: str | None = None, + api_key: str | None = None, + azure_ad_token: str | None = None, + azure_ad_token_provider: AsyncAzureADTokenProvider | None = None, + organization: str | None = None, + project: str | None = None, + base_url: str | None = None, + timeout: float | Timeout | None | NotGiven = NOT_GIVEN, + max_retries: int = DEFAULT_MAX_RETRIES, + default_headers: Mapping[str, str] | None = None, + default_query: Mapping[str, object] | None = None, + http_client: httpx.AsyncClient | None = None, + _strict_response_validation: bool = False, + ) -> None: + """Construct a new asynchronous azure openai client instance. + + This automatically infers the following arguments from their corresponding environment variables if they are not provided: + - `api_key` from `AZURE_OPENAI_API_KEY` + - `organization` from `OPENAI_ORG_ID` + - `project` from `OPENAI_PROJECT_ID` + - `azure_ad_token` from `AZURE_OPENAI_AD_TOKEN` + - `api_version` from `OPENAI_API_VERSION` + - `azure_endpoint` from `AZURE_OPENAI_ENDPOINT` + + Args: + azure_endpoint: Your Azure endpoint, including the resource, e.g. `https://example-resource.azure.openai.com/` + + azure_ad_token: Your Azure Active Directory token, https://www.microsoft.com/en-us/security/business/identity-access/microsoft-entra-id + + azure_ad_token_provider: A function that returns an Azure Active Directory token, will be invoked on every request. + + azure_deployment: A model deployment, if given sets the base client URL to include `/deployments/{azure_deployment}`. + Note: this means you won't be able to use non-deployment endpoints. Not supported with Assistants APIs. + """ + if api_key is None: + api_key = os.environ.get("AZURE_OPENAI_API_KEY") + + if azure_ad_token is None: + azure_ad_token = os.environ.get("AZURE_OPENAI_AD_TOKEN") + + if ( + api_key is None + and azure_ad_token is None + and azure_ad_token_provider is None + ): + raise OpenAIError( + "Missing credentials. Please pass one of `api_key`, `azure_ad_token`, `azure_ad_token_provider`, or the `AZURE_OPENAI_API_KEY` or `AZURE_OPENAI_AD_TOKEN` environment variables." + ) + + if api_version is None: + api_version = os.environ.get("OPENAI_API_VERSION") + + if api_version is None: + raise ValueError( + "Must provide either the `api_version` argument or the `OPENAI_API_VERSION` environment variable" + ) + + if default_query is None: + default_query = {"api-version": api_version} + else: + default_query = {**default_query, "api-version": api_version} + + if base_url is None: + if azure_endpoint is None: + azure_endpoint = os.environ.get("AZURE_OPENAI_ENDPOINT") + + if azure_endpoint is None: + raise ValueError( + "Must provide one of the `base_url` or `azure_endpoint` arguments, or the `AZURE_OPENAI_ENDPOINT` environment variable" + ) + + if azure_deployment is not None: + base_url = f"{azure_endpoint}/openai/deployments/{azure_deployment}" + else: + base_url = f"{azure_endpoint}/openai" + else: + if azure_endpoint is not None: + raise ValueError("base_url and azure_endpoint are mutually exclusive") + + if api_key is None: + # define a sentinel value to avoid any typing issues + api_key = API_KEY_SENTINEL + + super().__init__( + api_key=api_key, + organization=organization, + project=project, + base_url=base_url, + timeout=timeout, + max_retries=max_retries, + default_headers=default_headers, + default_query=default_query, + http_client=http_client, + _strict_response_validation=_strict_response_validation, + ) + self._api_version = api_version + self._azure_ad_token = azure_ad_token + self._azure_ad_token_provider = azure_ad_token_provider + + @override + def copy( + self, + *, + api_key: str | None = None, + organization: str | None = None, + project: str | None = None, + api_version: str | None = None, + azure_ad_token: str | None = None, + azure_ad_token_provider: AsyncAzureADTokenProvider | None = None, + base_url: str | httpx.URL | None = None, + timeout: float | Timeout | None | NotGiven = NOT_GIVEN, + http_client: httpx.AsyncClient | None = None, + max_retries: int | NotGiven = NOT_GIVEN, + default_headers: Mapping[str, str] | None = None, + set_default_headers: Mapping[str, str] | None = None, + default_query: Mapping[str, object] | None = None, + set_default_query: Mapping[str, object] | None = None, + _extra_kwargs: Mapping[str, Any] = {}, + ) -> Self: + """ + Create a new client instance re-using the same options given to the current client with optional overriding. + """ + return super().copy( + api_key=api_key, + organization=organization, + project=project, + base_url=base_url, + timeout=timeout, + http_client=http_client, + max_retries=max_retries, + default_headers=default_headers, + set_default_headers=set_default_headers, + default_query=default_query, + set_default_query=set_default_query, + _extra_kwargs={ + "api_version": api_version or self._api_version, + "azure_ad_token": azure_ad_token or self._azure_ad_token, + "azure_ad_token_provider": azure_ad_token_provider + or self._azure_ad_token_provider, + **_extra_kwargs, + }, + ) + + with_options = copy + + async def _get_azure_ad_token(self) -> str | None: + if self._azure_ad_token is not None: + return self._azure_ad_token + + provider = self._azure_ad_token_provider + if provider is not None: + token = provider() + if inspect.isawaitable(token): + token = await token + if not token or not isinstance(token, str): + raise ValueError( + f"Expected `azure_ad_token_provider` argument to return a string but it returned {token}", + ) + return token + + return None + + @override + async def _prepare_options(self, options: FinalRequestOptions) -> None: + headers: dict[str, str | Omit] = ( + {**options.headers} if is_given(options.headers) else {} + ) + options.headers = headers + + azure_ad_token = await self._get_azure_ad_token() + if azure_ad_token is not None: + if headers.get("Authorization") is None: + headers["Authorization"] = f"Bearer {azure_ad_token}" + elif self.api_key is not API_KEY_SENTINEL: + if headers.get("api-key") is None: + headers["api-key"] = self.api_key + else: + # should never be hit + raise ValueError("Unable to handle auth") + + return await super()._prepare_options(options) diff --git a/portkey_ai/_vendor/openai/lib/streaming/__init__.py b/portkey_ai/_vendor/openai/lib/streaming/__init__.py new file mode 100644 index 00000000..eb378d25 --- /dev/null +++ b/portkey_ai/_vendor/openai/lib/streaming/__init__.py @@ -0,0 +1,8 @@ +from ._assistants import ( + AssistantEventHandler as AssistantEventHandler, + AssistantEventHandlerT as AssistantEventHandlerT, + AssistantStreamManager as AssistantStreamManager, + AsyncAssistantEventHandler as AsyncAssistantEventHandler, + AsyncAssistantEventHandlerT as AsyncAssistantEventHandlerT, + AsyncAssistantStreamManager as AsyncAssistantStreamManager, +) diff --git a/portkey_ai/_vendor/openai/lib/streaming/_assistants.py b/portkey_ai/_vendor/openai/lib/streaming/_assistants.py new file mode 100644 index 00000000..15a22a6d --- /dev/null +++ b/portkey_ai/_vendor/openai/lib/streaming/_assistants.py @@ -0,0 +1,1114 @@ +from __future__ import annotations + +import asyncio +from types import TracebackType +from typing import ( + TYPE_CHECKING, + Any, + Generic, + TypeVar, + Callable, + Iterable, + Iterator, + cast, +) +from typing_extensions import Awaitable, AsyncIterable, AsyncIterator, assert_never + +import httpx + +from ..._utils import is_dict, is_list, consume_sync_iterator, consume_async_iterator +from ..._models import construct_type +from ..._streaming import Stream, AsyncStream +from ...types.beta import AssistantStreamEvent +from ...types.beta.threads import ( + Run, + Text, + Message, + ImageFile, + TextDelta, + MessageDelta, + MessageContent, + MessageContentDelta, +) +from ...types.beta.threads.runs import RunStep, ToolCall, RunStepDelta, ToolCallDelta + + +class AssistantEventHandler: + text_deltas: Iterable[str] + """Iterator over just the text deltas in the stream. + + This corresponds to the `thread.message.delta` event + in the API. + + ```py + for text in stream.text_deltas: + print(text, end="", flush=True) + print() + ``` + """ + + def __init__(self) -> None: + self._current_event: AssistantStreamEvent | None = None + self._current_message_content_index: int | None = None + self._current_message_content: MessageContent | None = None + self._current_tool_call_index: int | None = None + self._current_tool_call: ToolCall | None = None + self.__current_run_step_id: str | None = None + self.__current_run: Run | None = None + self.__run_step_snapshots: dict[str, RunStep] = {} + self.__message_snapshots: dict[str, Message] = {} + self.__current_message_snapshot: Message | None = None + + self.text_deltas = self.__text_deltas__() + self._iterator = self.__stream__() + self.__stream: Stream[AssistantStreamEvent] | None = None + + def _init(self, stream: Stream[AssistantStreamEvent]) -> None: + if self.__stream: + raise RuntimeError( + "A single event handler cannot be shared between multiple streams; You will need to construct a new event handler instance" + ) + + self.__stream = stream + + def __next__(self) -> AssistantStreamEvent: + return self._iterator.__next__() + + def __iter__(self) -> Iterator[AssistantStreamEvent]: + for item in self._iterator: + yield item + + @property + def current_event(self) -> AssistantStreamEvent | None: + return self._current_event + + @property + def current_run(self) -> Run | None: + return self.__current_run + + @property + def current_run_step_snapshot(self) -> RunStep | None: + if not self.__current_run_step_id: + return None + + return self.__run_step_snapshots[self.__current_run_step_id] + + @property + def current_message_snapshot(self) -> Message | None: + return self.__current_message_snapshot + + def close(self) -> None: + """ + Close the response and release the connection. + + Automatically called when the context manager exits. + """ + if self.__stream: + self.__stream.close() + + def until_done(self) -> None: + """Waits until the stream has been consumed""" + consume_sync_iterator(self) + + def get_final_run(self) -> Run: + """Wait for the stream to finish and returns the completed Run object""" + self.until_done() + + if not self.__current_run: + raise RuntimeError("No final run object found") + + return self.__current_run + + def get_final_run_steps(self) -> list[RunStep]: + """Wait for the stream to finish and returns the steps taken in this run""" + self.until_done() + + if not self.__run_step_snapshots: + raise RuntimeError("No run steps found") + + return [step for step in self.__run_step_snapshots.values()] + + def get_final_messages(self) -> list[Message]: + """Wait for the stream to finish and returns the messages emitted in this run""" + self.until_done() + + if not self.__message_snapshots: + raise RuntimeError("No messages found") + + return [message for message in self.__message_snapshots.values()] + + def __text_deltas__(self) -> Iterator[str]: + for event in self: + if event.event != "thread.message.delta": + continue + + for content_delta in event.data.delta.content or []: + if ( + content_delta.type == "text" + and content_delta.text + and content_delta.text.value + ): + yield content_delta.text.value + + # event handlers + + def on_end(self) -> None: + """Fires when the stream has finished. + + This happens if the stream is read to completion + or if an exception occurs during iteration. + """ + + def on_event(self, event: AssistantStreamEvent) -> None: + """Callback that is fired for every Server-Sent-Event""" + + def on_run_step_created(self, run_step: RunStep) -> None: + """Callback that is fired when a run step is created""" + + def on_run_step_delta(self, delta: RunStepDelta, snapshot: RunStep) -> None: + """Callback that is fired whenever a run step delta is returned from the API + + The first argument is just the delta as sent by the API and the second argument + is the accumulated snapshot of the run step. For example, a tool calls event may + look like this: + + # delta + tool_calls=[ + RunStepDeltaToolCallsCodeInterpreter( + index=0, + type='code_interpreter', + id=None, + code_interpreter=CodeInterpreter(input=' sympy', outputs=None) + ) + ] + # snapshot + tool_calls=[ + CodeToolCall( + id='call_wKayJlcYV12NiadiZuJXxcfx', + code_interpreter=CodeInterpreter(input='from sympy', outputs=[]), + type='code_interpreter', + index=0 + ) + ], + """ + + def on_run_step_done(self, run_step: RunStep) -> None: + """Callback that is fired when a run step is completed""" + + def on_tool_call_created(self, tool_call: ToolCall) -> None: + """Callback that is fired when a tool call is created""" + + def on_tool_call_delta(self, delta: ToolCallDelta, snapshot: ToolCall) -> None: + """Callback that is fired when a tool call delta is encountered""" + + def on_tool_call_done(self, tool_call: ToolCall) -> None: + """Callback that is fired when a tool call delta is encountered""" + + def on_exception(self, exception: Exception) -> None: + """Fired whenever an exception happens during streaming""" + + def on_timeout(self) -> None: + """Fires if the request times out""" + + def on_message_created(self, message: Message) -> None: + """Callback that is fired when a message is created""" + + def on_message_delta(self, delta: MessageDelta, snapshot: Message) -> None: + """Callback that is fired whenever a message delta is returned from the API + + The first argument is just the delta as sent by the API and the second argument + is the accumulated snapshot of the message. For example, a text content event may + look like this: + + # delta + MessageDeltaText( + index=0, + type='text', + text=Text( + value=' Jane' + ), + ) + # snapshot + MessageContentText( + index=0, + type='text', + text=Text( + value='Certainly, Jane' + ), + ) + """ + + def on_message_done(self, message: Message) -> None: + """Callback that is fired when a message is completed""" + + def on_text_created(self, text: Text) -> None: + """Callback that is fired when a text content block is created""" + + def on_text_delta(self, delta: TextDelta, snapshot: Text) -> None: + """Callback that is fired whenever a text content delta is returned + by the API. + + The first argument is just the delta as sent by the API and the second argument + is the accumulated snapshot of the text. For example: + + on_text_delta(TextDelta(value="The"), Text(value="The")), + on_text_delta(TextDelta(value=" solution"), Text(value="The solution")), + on_text_delta(TextDelta(value=" to"), Text(value="The solution to")), + on_text_delta(TextDelta(value=" the"), Text(value="The solution to the")), + on_text_delta(TextDelta(value=" equation"), Text(value="The solution to the equivalent")), + """ + + def on_text_done(self, text: Text) -> None: + """Callback that is fired when a text content block is finished""" + + def on_image_file_done(self, image_file: ImageFile) -> None: + """Callback that is fired when an image file block is finished""" + + def _emit_sse_event(self, event: AssistantStreamEvent) -> None: + self._current_event = event + self.on_event(event) + + self.__current_message_snapshot, new_content = accumulate_event( + event=event, + current_message_snapshot=self.__current_message_snapshot, + ) + if self.__current_message_snapshot is not None: + self.__message_snapshots[ + self.__current_message_snapshot.id + ] = self.__current_message_snapshot + + accumulate_run_step( + event=event, + run_step_snapshots=self.__run_step_snapshots, + ) + + for content_delta in new_content: + assert self.__current_message_snapshot is not None + + block = self.__current_message_snapshot.content[content_delta.index] + if block.type == "text": + self.on_text_created(block.text) + + if ( + event.event == "thread.run.completed" + or event.event == "thread.run.cancelled" + or event.event == "thread.run.expired" + or event.event == "thread.run.failed" + or event.event == "thread.run.requires_action" + ): + self.__current_run = event.data + if self._current_tool_call: + self.on_tool_call_done(self._current_tool_call) + elif ( + event.event == "thread.run.created" + or event.event == "thread.run.in_progress" + or event.event == "thread.run.cancelling" + or event.event == "thread.run.queued" + ): + self.__current_run = event.data + elif event.event == "thread.message.created": + self.on_message_created(event.data) + elif event.event == "thread.message.delta": + snapshot = self.__current_message_snapshot + assert snapshot is not None + + message_delta = event.data.delta + if message_delta.content is not None: + for content_delta in message_delta.content: + if content_delta.type == "text" and content_delta.text: + snapshot_content = snapshot.content[content_delta.index] + assert snapshot_content.type == "text" + self.on_text_delta(content_delta.text, snapshot_content.text) + + # If the delta is for a new message content: + # - emit on_text_done/on_image_file_done for the previous message content + # - emit on_text_created/on_image_created for the new message content + if content_delta.index != self._current_message_content_index: + if self._current_message_content is not None: + if self._current_message_content.type == "text": + self.on_text_done(self._current_message_content.text) + elif self._current_message_content.type == "image_file": + self.on_image_file_done( + self._current_message_content.image_file + ) + + self._current_message_content_index = content_delta.index + self._current_message_content = snapshot.content[ + content_delta.index + ] + + # Update the current_message_content (delta event is correctly emitted already) + self._current_message_content = snapshot.content[ + content_delta.index + ] + + self.on_message_delta(event.data.delta, snapshot) + elif ( + event.event == "thread.message.completed" + or event.event == "thread.message.incomplete" + ): + self.__current_message_snapshot = event.data + self.__message_snapshots[event.data.id] = event.data + + if self._current_message_content_index is not None: + content = event.data.content[self._current_message_content_index] + if content.type == "text": + self.on_text_done(content.text) + elif content.type == "image_file": + self.on_image_file_done(content.image_file) + + self.on_message_done(event.data) + elif event.event == "thread.run.step.created": + self.__current_run_step_id = event.data.id + self.on_run_step_created(event.data) + elif event.event == "thread.run.step.in_progress": + self.__current_run_step_id = event.data.id + elif event.event == "thread.run.step.delta": + step_snapshot = self.__run_step_snapshots[event.data.id] + + run_step_delta = event.data.delta + if ( + run_step_delta.step_details + and run_step_delta.step_details.type == "tool_calls" + and run_step_delta.step_details.tool_calls is not None + ): + assert step_snapshot.step_details.type == "tool_calls" + for tool_call_delta in run_step_delta.step_details.tool_calls: + if tool_call_delta.index == self._current_tool_call_index: + self.on_tool_call_delta( + tool_call_delta, + step_snapshot.step_details.tool_calls[ + tool_call_delta.index + ], + ) + + # If the delta is for a new tool call: + # - emit on_tool_call_done for the previous tool_call + # - emit on_tool_call_created for the new tool_call + if tool_call_delta.index != self._current_tool_call_index: + if self._current_tool_call is not None: + self.on_tool_call_done(self._current_tool_call) + + self._current_tool_call_index = tool_call_delta.index + self._current_tool_call = step_snapshot.step_details.tool_calls[ + tool_call_delta.index + ] + self.on_tool_call_created(self._current_tool_call) + + # Update the current_tool_call (delta event is correctly emitted already) + self._current_tool_call = step_snapshot.step_details.tool_calls[ + tool_call_delta.index + ] + + self.on_run_step_delta( + event.data.delta, + step_snapshot, + ) + elif ( + event.event == "thread.run.step.completed" + or event.event == "thread.run.step.cancelled" + or event.event == "thread.run.step.expired" + or event.event == "thread.run.step.failed" + ): + if self._current_tool_call: + self.on_tool_call_done(self._current_tool_call) + + self.on_run_step_done(event.data) + self.__current_run_step_id = None + elif ( + event.event == "thread.created" + or event.event == "thread.message.in_progress" + or event.event == "error" + ): + # currently no special handling + ... + else: + # we only want to error at build-time + if TYPE_CHECKING: # type: ignore[unreachable] + assert_never(event) + + self._current_event = None + + def __stream__(self) -> Iterator[AssistantStreamEvent]: + stream = self.__stream + if not stream: + raise RuntimeError("Stream has not been started yet") + + try: + for event in stream: + self._emit_sse_event(event) + + yield event + except (httpx.TimeoutException, asyncio.TimeoutError) as exc: + self.on_timeout() + self.on_exception(exc) + raise + except Exception as exc: + self.on_exception(exc) + raise + finally: + self.on_end() + + +AssistantEventHandlerT = TypeVar("AssistantEventHandlerT", bound=AssistantEventHandler) + + +class AssistantStreamManager(Generic[AssistantEventHandlerT]): + """Wrapper over AssistantStreamEventHandler that is returned by `.stream()` + so that a context manager can be used. + + ```py + with client.threads.create_and_run_stream(...) as stream: + for event in stream: + ... + ``` + """ + + def __init__( + self, + api_request: Callable[[], Stream[AssistantStreamEvent]], + *, + event_handler: AssistantEventHandlerT, + ) -> None: + self.__stream: Stream[AssistantStreamEvent] | None = None + self.__event_handler = event_handler + self.__api_request = api_request + + def __enter__(self) -> AssistantEventHandlerT: + self.__stream = self.__api_request() + self.__event_handler._init(self.__stream) + return self.__event_handler + + def __exit__( + self, + exc_type: type[BaseException] | None, + exc: BaseException | None, + exc_tb: TracebackType | None, + ) -> None: + if self.__stream is not None: + self.__stream.close() + + +class AsyncAssistantEventHandler: + text_deltas: AsyncIterable[str] + """Iterator over just the text deltas in the stream. + + This corresponds to the `thread.message.delta` event + in the API. + + ```py + async for text in stream.text_deltas: + print(text, end="", flush=True) + print() + ``` + """ + + def __init__(self) -> None: + self._current_event: AssistantStreamEvent | None = None + self._current_message_content_index: int | None = None + self._current_message_content: MessageContent | None = None + self._current_tool_call_index: int | None = None + self._current_tool_call: ToolCall | None = None + self.__current_run_step_id: str | None = None + self.__current_run: Run | None = None + self.__run_step_snapshots: dict[str, RunStep] = {} + self.__message_snapshots: dict[str, Message] = {} + self.__current_message_snapshot: Message | None = None + + self.text_deltas = self.__text_deltas__() + self._iterator = self.__stream__() + self.__stream: AsyncStream[AssistantStreamEvent] | None = None + + def _init(self, stream: AsyncStream[AssistantStreamEvent]) -> None: + if self.__stream: + raise RuntimeError( + "A single event handler cannot be shared between multiple streams; You will need to construct a new event handler instance" + ) + + self.__stream = stream + + async def __anext__(self) -> AssistantStreamEvent: + return await self._iterator.__anext__() + + async def __aiter__(self) -> AsyncIterator[AssistantStreamEvent]: + async for item in self._iterator: + yield item + + async def close(self) -> None: + """ + Close the response and release the connection. + + Automatically called when the context manager exits. + """ + if self.__stream: + await self.__stream.close() + + @property + def current_event(self) -> AssistantStreamEvent | None: + return self._current_event + + @property + def current_run(self) -> Run | None: + return self.__current_run + + @property + def current_run_step_snapshot(self) -> RunStep | None: + if not self.__current_run_step_id: + return None + + return self.__run_step_snapshots[self.__current_run_step_id] + + @property + def current_message_snapshot(self) -> Message | None: + return self.__current_message_snapshot + + async def until_done(self) -> None: + """Waits until the stream has been consumed""" + await consume_async_iterator(self) + + async def get_final_run(self) -> Run: + """Wait for the stream to finish and returns the completed Run object""" + await self.until_done() + + if not self.__current_run: + raise RuntimeError("No final run object found") + + return self.__current_run + + async def get_final_run_steps(self) -> list[RunStep]: + """Wait for the stream to finish and returns the steps taken in this run""" + await self.until_done() + + if not self.__run_step_snapshots: + raise RuntimeError("No run steps found") + + return [step for step in self.__run_step_snapshots.values()] + + async def get_final_messages(self) -> list[Message]: + """Wait for the stream to finish and returns the messages emitted in this run""" + await self.until_done() + + if not self.__message_snapshots: + raise RuntimeError("No messages found") + + return [message for message in self.__message_snapshots.values()] + + async def __text_deltas__(self) -> AsyncIterator[str]: + async for event in self: + if event.event != "thread.message.delta": + continue + + for content_delta in event.data.delta.content or []: + if ( + content_delta.type == "text" + and content_delta.text + and content_delta.text.value + ): + yield content_delta.text.value + + # event handlers + + async def on_end(self) -> None: + """Fires when the stream has finished. + + This happens if the stream is read to completion + or if an exception occurs during iteration. + """ + + async def on_event(self, event: AssistantStreamEvent) -> None: + """Callback that is fired for every Server-Sent-Event""" + + async def on_run_step_created(self, run_step: RunStep) -> None: + """Callback that is fired when a run step is created""" + + async def on_run_step_delta(self, delta: RunStepDelta, snapshot: RunStep) -> None: + """Callback that is fired whenever a run step delta is returned from the API + + The first argument is just the delta as sent by the API and the second argument + is the accumulated snapshot of the run step. For example, a tool calls event may + look like this: + + # delta + tool_calls=[ + RunStepDeltaToolCallsCodeInterpreter( + index=0, + type='code_interpreter', + id=None, + code_interpreter=CodeInterpreter(input=' sympy', outputs=None) + ) + ] + # snapshot + tool_calls=[ + CodeToolCall( + id='call_wKayJlcYV12NiadiZuJXxcfx', + code_interpreter=CodeInterpreter(input='from sympy', outputs=[]), + type='code_interpreter', + index=0 + ) + ], + """ + + async def on_run_step_done(self, run_step: RunStep) -> None: + """Callback that is fired when a run step is completed""" + + async def on_tool_call_created(self, tool_call: ToolCall) -> None: + """Callback that is fired when a tool call is created""" + + async def on_tool_call_delta( + self, delta: ToolCallDelta, snapshot: ToolCall + ) -> None: + """Callback that is fired when a tool call delta is encountered""" + + async def on_tool_call_done(self, tool_call: ToolCall) -> None: + """Callback that is fired when a tool call delta is encountered""" + + async def on_exception(self, exception: Exception) -> None: + """Fired whenever an exception happens during streaming""" + + async def on_timeout(self) -> None: + """Fires if the request times out""" + + async def on_message_created(self, message: Message) -> None: + """Callback that is fired when a message is created""" + + async def on_message_delta(self, delta: MessageDelta, snapshot: Message) -> None: + """Callback that is fired whenever a message delta is returned from the API + + The first argument is just the delta as sent by the API and the second argument + is the accumulated snapshot of the message. For example, a text content event may + look like this: + + # delta + MessageDeltaText( + index=0, + type='text', + text=Text( + value=' Jane' + ), + ) + # snapshot + MessageContentText( + index=0, + type='text', + text=Text( + value='Certainly, Jane' + ), + ) + """ + + async def on_message_done(self, message: Message) -> None: + """Callback that is fired when a message is completed""" + + async def on_text_created(self, text: Text) -> None: + """Callback that is fired when a text content block is created""" + + async def on_text_delta(self, delta: TextDelta, snapshot: Text) -> None: + """Callback that is fired whenever a text content delta is returned + by the API. + + The first argument is just the delta as sent by the API and the second argument + is the accumulated snapshot of the text. For example: + + on_text_delta(TextDelta(value="The"), Text(value="The")), + on_text_delta(TextDelta(value=" solution"), Text(value="The solution")), + on_text_delta(TextDelta(value=" to"), Text(value="The solution to")), + on_text_delta(TextDelta(value=" the"), Text(value="The solution to the")), + on_text_delta(TextDelta(value=" equation"), Text(value="The solution to the equivalent")), + """ + + async def on_text_done(self, text: Text) -> None: + """Callback that is fired when a text content block is finished""" + + async def on_image_file_done(self, image_file: ImageFile) -> None: + """Callback that is fired when an image file block is finished""" + + async def _emit_sse_event(self, event: AssistantStreamEvent) -> None: + self._current_event = event + await self.on_event(event) + + self.__current_message_snapshot, new_content = accumulate_event( + event=event, + current_message_snapshot=self.__current_message_snapshot, + ) + if self.__current_message_snapshot is not None: + self.__message_snapshots[ + self.__current_message_snapshot.id + ] = self.__current_message_snapshot + + accumulate_run_step( + event=event, + run_step_snapshots=self.__run_step_snapshots, + ) + + for content_delta in new_content: + assert self.__current_message_snapshot is not None + + block = self.__current_message_snapshot.content[content_delta.index] + if block.type == "text": + await self.on_text_created(block.text) + + if ( + event.event == "thread.run.completed" + or event.event == "thread.run.cancelled" + or event.event == "thread.run.expired" + or event.event == "thread.run.failed" + or event.event == "thread.run.requires_action" + ): + self.__current_run = event.data + if self._current_tool_call: + await self.on_tool_call_done(self._current_tool_call) + elif ( + event.event == "thread.run.created" + or event.event == "thread.run.in_progress" + or event.event == "thread.run.cancelling" + or event.event == "thread.run.queued" + ): + self.__current_run = event.data + elif event.event == "thread.message.created": + await self.on_message_created(event.data) + elif event.event == "thread.message.delta": + snapshot = self.__current_message_snapshot + assert snapshot is not None + + message_delta = event.data.delta + if message_delta.content is not None: + for content_delta in message_delta.content: + if content_delta.type == "text" and content_delta.text: + snapshot_content = snapshot.content[content_delta.index] + assert snapshot_content.type == "text" + await self.on_text_delta( + content_delta.text, snapshot_content.text + ) + + # If the delta is for a new message content: + # - emit on_text_done/on_image_file_done for the previous message content + # - emit on_text_created/on_image_created for the new message content + if content_delta.index != self._current_message_content_index: + if self._current_message_content is not None: + if self._current_message_content.type == "text": + await self.on_text_done( + self._current_message_content.text + ) + elif self._current_message_content.type == "image_file": + await self.on_image_file_done( + self._current_message_content.image_file + ) + + self._current_message_content_index = content_delta.index + self._current_message_content = snapshot.content[ + content_delta.index + ] + + # Update the current_message_content (delta event is correctly emitted already) + self._current_message_content = snapshot.content[ + content_delta.index + ] + + await self.on_message_delta(event.data.delta, snapshot) + elif ( + event.event == "thread.message.completed" + or event.event == "thread.message.incomplete" + ): + self.__current_message_snapshot = event.data + self.__message_snapshots[event.data.id] = event.data + + if self._current_message_content_index is not None: + content = event.data.content[self._current_message_content_index] + if content.type == "text": + await self.on_text_done(content.text) + elif content.type == "image_file": + await self.on_image_file_done(content.image_file) + + await self.on_message_done(event.data) + elif event.event == "thread.run.step.created": + self.__current_run_step_id = event.data.id + await self.on_run_step_created(event.data) + elif event.event == "thread.run.step.in_progress": + self.__current_run_step_id = event.data.id + elif event.event == "thread.run.step.delta": + step_snapshot = self.__run_step_snapshots[event.data.id] + + run_step_delta = event.data.delta + if ( + run_step_delta.step_details + and run_step_delta.step_details.type == "tool_calls" + and run_step_delta.step_details.tool_calls is not None + ): + assert step_snapshot.step_details.type == "tool_calls" + for tool_call_delta in run_step_delta.step_details.tool_calls: + if tool_call_delta.index == self._current_tool_call_index: + await self.on_tool_call_delta( + tool_call_delta, + step_snapshot.step_details.tool_calls[ + tool_call_delta.index + ], + ) + + # If the delta is for a new tool call: + # - emit on_tool_call_done for the previous tool_call + # - emit on_tool_call_created for the new tool_call + if tool_call_delta.index != self._current_tool_call_index: + if self._current_tool_call is not None: + await self.on_tool_call_done(self._current_tool_call) + + self._current_tool_call_index = tool_call_delta.index + self._current_tool_call = step_snapshot.step_details.tool_calls[ + tool_call_delta.index + ] + await self.on_tool_call_created(self._current_tool_call) + + # Update the current_tool_call (delta event is correctly emitted already) + self._current_tool_call = step_snapshot.step_details.tool_calls[ + tool_call_delta.index + ] + + await self.on_run_step_delta( + event.data.delta, + step_snapshot, + ) + elif ( + event.event == "thread.run.step.completed" + or event.event == "thread.run.step.cancelled" + or event.event == "thread.run.step.expired" + or event.event == "thread.run.step.failed" + ): + if self._current_tool_call: + await self.on_tool_call_done(self._current_tool_call) + + await self.on_run_step_done(event.data) + self.__current_run_step_id = None + elif ( + event.event == "thread.created" + or event.event == "thread.message.in_progress" + or event.event == "error" + ): + # currently no special handling + ... + else: + # we only want to error at build-time + if TYPE_CHECKING: # type: ignore[unreachable] + assert_never(event) + + self._current_event = None + + async def __stream__(self) -> AsyncIterator[AssistantStreamEvent]: + stream = self.__stream + if not stream: + raise RuntimeError("Stream has not been started yet") + + try: + async for event in stream: + await self._emit_sse_event(event) + + yield event + except (httpx.TimeoutException, asyncio.TimeoutError) as exc: + await self.on_timeout() + await self.on_exception(exc) + raise + except Exception as exc: + await self.on_exception(exc) + raise + finally: + await self.on_end() + + +AsyncAssistantEventHandlerT = TypeVar( + "AsyncAssistantEventHandlerT", bound=AsyncAssistantEventHandler +) + + +class AsyncAssistantStreamManager(Generic[AsyncAssistantEventHandlerT]): + """Wrapper over AsyncAssistantStreamEventHandler that is returned by `.stream()` + so that an async context manager can be used without `await`ing the + original client call. + + ```py + async with client.threads.create_and_run_stream(...) as stream: + async for event in stream: + ... + ``` + """ + + def __init__( + self, + api_request: Awaitable[AsyncStream[AssistantStreamEvent]], + *, + event_handler: AsyncAssistantEventHandlerT, + ) -> None: + self.__stream: AsyncStream[AssistantStreamEvent] | None = None + self.__event_handler = event_handler + self.__api_request = api_request + + async def __aenter__(self) -> AsyncAssistantEventHandlerT: + self.__stream = await self.__api_request + self.__event_handler._init(self.__stream) + return self.__event_handler + + async def __aexit__( + self, + exc_type: type[BaseException] | None, + exc: BaseException | None, + exc_tb: TracebackType | None, + ) -> None: + if self.__stream is not None: + await self.__stream.close() + + +def accumulate_run_step( + *, + event: AssistantStreamEvent, + run_step_snapshots: dict[str, RunStep], +) -> None: + if event.event == "thread.run.step.created": + run_step_snapshots[event.data.id] = event.data + return + + if event.event == "thread.run.step.delta": + data = event.data + snapshot = run_step_snapshots[data.id] + + if data.delta: + merged = accumulate_delta( + cast( + "dict[object, object]", + snapshot.model_dump(exclude_unset=True), + ), + cast( + "dict[object, object]", + data.delta.model_dump(exclude_unset=True), + ), + ) + run_step_snapshots[snapshot.id] = cast( + RunStep, construct_type(type_=RunStep, value=merged) + ) + + return None + + +def accumulate_event( + *, + event: AssistantStreamEvent, + current_message_snapshot: Message | None, +) -> tuple[Message | None, list[MessageContentDelta]]: + """Returns a tuple of message snapshot and newly created text message deltas""" + if event.event == "thread.message.created": + return event.data, [] + + new_content: list[MessageContentDelta] = [] + + if event.event != "thread.message.delta": + return current_message_snapshot, [] + + if not current_message_snapshot: + raise RuntimeError("Encountered a message delta with no previous snapshot") + + data = event.data + if data.delta.content: + for content_delta in data.delta.content: + try: + block = current_message_snapshot.content[content_delta.index] + except IndexError: + current_message_snapshot.content.insert( + content_delta.index, + cast( + MessageContent, + construct_type( + # mypy doesn't allow Content for some reason + type_=cast(Any, MessageContent), + value=content_delta.model_dump(exclude_unset=True), + ), + ), + ) + new_content.append(content_delta) + else: + merged = accumulate_delta( + cast( + "dict[object, object]", + block.model_dump(exclude_unset=True), + ), + cast( + "dict[object, object]", + content_delta.model_dump(exclude_unset=True), + ), + ) + current_message_snapshot.content[content_delta.index] = cast( + MessageContent, + construct_type( + # mypy doesn't allow Content for some reason + type_=cast(Any, MessageContent), + value=merged, + ), + ) + + return current_message_snapshot, new_content + + +def accumulate_delta( + acc: dict[object, object], delta: dict[object, object] +) -> dict[object, object]: + for key, delta_value in delta.items(): + if key not in acc: + acc[key] = delta_value + continue + + acc_value = acc[key] + if acc_value is None: + acc[key] = delta_value + continue + + # the `index` property is used in arrays of objects so it should + # not be accumulated like other values e.g. + # [{'foo': 'bar', 'index': 0}] + # + # the same applies to `type` properties as they're used for + # discriminated unions + if key == "index" or key == "type": + acc[key] = delta_value + continue + + if isinstance(acc_value, str) and isinstance(delta_value, str): + acc_value += delta_value + elif isinstance(acc_value, (int, float)) and isinstance( + delta_value, (int, float) + ): + acc_value += delta_value + elif is_dict(acc_value) and is_dict(delta_value): + acc_value = accumulate_delta(acc_value, delta_value) + elif is_list(acc_value) and is_list(delta_value): + # for lists of non-dictionary items we'll only ever get new entries + # in the array, existing entries will never be changed + if all(isinstance(x, (str, int, float)) for x in acc_value): + acc_value.extend(delta_value) + continue + + for delta_entry in delta_value: + if not is_dict(delta_entry): + raise TypeError( + f"Unexpected list delta entry is not a dictionary: {delta_entry}" + ) + + try: + index = delta_entry["index"] + except KeyError as exc: + raise RuntimeError( + f"Expected list delta entry to have an `index` key; {delta_entry}" + ) from exc + + if not isinstance(index, int): + raise TypeError( + f"Unexpected, list delta entry `index` value is not an integer; {index}" + ) + + try: + acc_entry = acc_value[index] + except IndexError: + acc_value.insert(index, delta_entry) + else: + if not is_dict(acc_entry): + raise TypeError("not handled yet") + + acc_value[index] = accumulate_delta(acc_entry, delta_entry) + + acc[key] = acc_value + + return acc diff --git a/setup.cfg b/setup.cfg index 8c280448..79f80d65 100644 --- a/setup.cfg +++ b/setup.cfg @@ -23,7 +23,11 @@ install_requires = mypy>=0.991,<2.0 typing_extensions>=4.7.1,<5.0 pydantic>=1.10.12 - openai>=1.0,<2.0 + anyio>=3.5.0,<5 + distro>=1.7.0,<2 + sniffio + cached-property + tqdm>4 [options.entry_points] console_scripts = @@ -42,7 +46,6 @@ dev = python-dotenv==1.0.0 ruff==0.0.292 pytest-asyncio==0.23.5 - openai>=1.0,<2.0 [mypy] ignore_missing_imports = true