# ---------------------------------------------------------------------------- # # Welcome to Baml! To use this generated code, please run the following: # # $ pip install baml # # ---------------------------------------------------------------------------- # This file was generated by BAML: please do not edit it. Instead, edit the # BAML files and re-generate this code using: baml-cli generate # baml-cli is available with the baml package. import typing import typing_extensions import baml_py from . import stream_types, types, type_builder from .parser import LlmResponseParser, LlmStreamParser from .runtime import DoNotUseDirectlyCallManager, BamlCallOptions from .globals import DO_NOT_USE_DIRECTLY_UNLESS_YOU_KNOW_WHAT_YOURE_DOING_RUNTIME as __runtime__ class BamlSyncClient: __options: DoNotUseDirectlyCallManager __stream_client: "BamlStreamClient" __http_request: "BamlHttpRequestClient" __http_stream_request: "BamlHttpStreamRequestClient" __llm_response_parser: LlmResponseParser __llm_stream_parser: LlmStreamParser def __init__(self, options: DoNotUseDirectlyCallManager): self.__options = options self.__stream_client = BamlStreamClient(options) self.__http_request = BamlHttpRequestClient(options) self.__http_stream_request = BamlHttpStreamRequestClient(options) self.__llm_response_parser = LlmResponseParser(options) self.__llm_stream_parser = LlmStreamParser(options) def __getstate__(self): # Return state needed for pickling return {"options": self.__options} def __setstate__(self, state): # Restore state from pickling self.__options = state["options"] self.__stream_client = BamlStreamClient(self.__options) self.__http_request = BamlHttpRequestClient(self.__options) self.__http_stream_request = BamlHttpStreamRequestClient(self.__options) self.__llm_response_parser = LlmResponseParser(self.__options) self.__llm_stream_parser = LlmStreamParser(self.__options) def with_options(self, tb: typing.Optional[type_builder.TypeBuilder] = None, client_registry: typing.Optional[baml_py.baml_py.ClientRegistry] = None, client: typing.Optional[str] = None, collector: typing.Optional[typing.Union[baml_py.baml_py.Collector, typing.List[baml_py.baml_py.Collector]]] = None, env: typing.Optional[typing.Dict[str, typing.Optional[str]]] = None, tags: typing.Optional[typing.Dict[str, str]] = None, on_tick: typing.Optional[typing.Callable[[str, baml_py.baml_py.FunctionLog], None]] = None, ) -> "BamlSyncClient": options: BamlCallOptions = {} if tb is not None: options["tb"] = tb if client_registry is not None: options["client_registry"] = client_registry if client is not None: options["client"] = client if collector is not None: options["collector"] = collector if env is not None: options["env"] = env if tags is not None: options["tags"] = tags if on_tick is not None: options["on_tick"] = on_tick return BamlSyncClient(self.__options.merge_options(options)) @property def stream(self): return self.__stream_client @property def request(self): return self.__http_request @property def stream_request(self): return self.__http_stream_request @property def parse(self): return self.__llm_response_parser @property def parse_stream(self): return self.__llm_stream_parser def FilterItems(self, items: typing.List["types.FilterItem"], baml_options: BamlCallOptions = {}, ) -> typing.List["types.Verdict"]: # Check if on_tick is provided if 'on_tick' in baml_options: __stream__ = self.stream.FilterItems(items=items, baml_options=baml_options) return __stream__.get_final_response() else: # Original non-streaming code __result__ = self.__options.merge_options(baml_options).call_function_sync(function_name="FilterItems", args={ "items": items, }) return typing.cast(typing.List["types.Verdict"], __result__.cast_to(types, types, stream_types, False, __runtime__)) def FilterScoreItems(self, items: typing.List["types.ScoreItem"], baml_options: BamlCallOptions = {}, ) -> typing.List["types.RankedItem"]: # Check if on_tick is provided if 'on_tick' in baml_options: __stream__ = self.stream.FilterScoreItems(items=items, baml_options=baml_options) return __stream__.get_final_response() else: # Original non-streaming code __result__ = self.__options.merge_options(baml_options).call_function_sync(function_name="FilterScoreItems", args={ "items": items, }) return typing.cast(typing.List["types.RankedItem"], __result__.cast_to(types, types, stream_types, False, __runtime__)) def GenerateOutline(self, date: str,stories: typing.List["types.StoryItem"], baml_options: BamlCallOptions = {}, ) -> types.EpisodeOutline: # Check if on_tick is provided if 'on_tick' in baml_options: __stream__ = self.stream.GenerateOutline(date=date,stories=stories, baml_options=baml_options) return __stream__.get_final_response() else: # Original non-streaming code __result__ = self.__options.merge_options(baml_options).call_function_sync(function_name="GenerateOutline", args={ "date": date,"stories": stories, }) return typing.cast(types.EpisodeOutline, __result__.cast_to(types, types, stream_types, False, __runtime__)) def GenerateScriptSegment(self, date: str,story_title: str,content_chunk: str,previous_context: typing.Optional[str],segment_type: str,outline_context: typing.Optional[str] = None, baml_options: BamlCallOptions = {}, ) -> typing.List["types.ScriptLine"]: # Check if on_tick is provided if 'on_tick' in baml_options: __stream__ = self.stream.GenerateScriptSegment(date=date,story_title=story_title,content_chunk=content_chunk,previous_context=previous_context,segment_type=segment_type,outline_context=outline_context, baml_options=baml_options) return __stream__.get_final_response() else: # Original non-streaming code __result__ = self.__options.merge_options(baml_options).call_function_sync(function_name="GenerateScriptSegment", args={ "date": date,"story_title": story_title,"content_chunk": content_chunk,"previous_context": previous_context,"segment_type": segment_type,"outline_context": outline_context, }) return typing.cast(typing.List["types.ScriptLine"], __result__.cast_to(types, types, stream_types, False, __runtime__)) def ScoreItems(self, items: typing.List["types.ScoreItem"], baml_options: BamlCallOptions = {}, ) -> typing.List["types.ScoredItem"]: # Check if on_tick is provided if 'on_tick' in baml_options: __stream__ = self.stream.ScoreItems(items=items, baml_options=baml_options) return __stream__.get_final_response() else: # Original non-streaming code __result__ = self.__options.merge_options(baml_options).call_function_sync(function_name="ScoreItems", args={ "items": items, }) return typing.cast(typing.List["types.ScoredItem"], __result__.cast_to(types, types, stream_types, False, __runtime__)) class BamlStreamClient: __options: DoNotUseDirectlyCallManager def __init__(self, options: DoNotUseDirectlyCallManager): self.__options = options def FilterItems(self, items: typing.List["types.FilterItem"], baml_options: BamlCallOptions = {}, ) -> baml_py.BamlSyncStream[typing.List["stream_types.Verdict"], typing.List["types.Verdict"]]: __ctx__, __result__ = self.__options.merge_options(baml_options).create_sync_stream(function_name="FilterItems", args={ "items": items, }) return baml_py.BamlSyncStream[typing.List["stream_types.Verdict"], typing.List["types.Verdict"]]( __result__, lambda x: typing.cast(typing.List["stream_types.Verdict"], x.cast_to(types, types, stream_types, True, __runtime__)), lambda x: typing.cast(typing.List["types.Verdict"], x.cast_to(types, types, stream_types, False, __runtime__)), __ctx__, ) def FilterScoreItems(self, items: typing.List["types.ScoreItem"], baml_options: BamlCallOptions = {}, ) -> baml_py.BamlSyncStream[typing.List["stream_types.RankedItem"], typing.List["types.RankedItem"]]: __ctx__, __result__ = self.__options.merge_options(baml_options).create_sync_stream(function_name="FilterScoreItems", args={ "items": items, }) return baml_py.BamlSyncStream[typing.List["stream_types.RankedItem"], typing.List["types.RankedItem"]]( __result__, lambda x: typing.cast(typing.List["stream_types.RankedItem"], x.cast_to(types, types, stream_types, True, __runtime__)), lambda x: typing.cast(typing.List["types.RankedItem"], x.cast_to(types, types, stream_types, False, __runtime__)), __ctx__, ) def GenerateOutline(self, date: str,stories: typing.List["types.StoryItem"], baml_options: BamlCallOptions = {}, ) -> baml_py.BamlSyncStream[stream_types.EpisodeOutline, types.EpisodeOutline]: __ctx__, __result__ = self.__options.merge_options(baml_options).create_sync_stream(function_name="GenerateOutline", args={ "date": date,"stories": stories, }) return baml_py.BamlSyncStream[stream_types.EpisodeOutline, types.EpisodeOutline]( __result__, lambda x: typing.cast(stream_types.EpisodeOutline, x.cast_to(types, types, stream_types, True, __runtime__)), lambda x: typing.cast(types.EpisodeOutline, x.cast_to(types, types, stream_types, False, __runtime__)), __ctx__, ) def GenerateScriptSegment(self, date: str,story_title: str,content_chunk: str,previous_context: typing.Optional[str],segment_type: str,outline_context: typing.Optional[str] = None, baml_options: BamlCallOptions = {}, ) -> baml_py.BamlSyncStream[typing.List["stream_types.ScriptLine"], typing.List["types.ScriptLine"]]: __ctx__, __result__ = self.__options.merge_options(baml_options).create_sync_stream(function_name="GenerateScriptSegment", args={ "date": date,"story_title": story_title,"content_chunk": content_chunk,"previous_context": previous_context,"segment_type": segment_type,"outline_context": outline_context, }) return baml_py.BamlSyncStream[typing.List["stream_types.ScriptLine"], typing.List["types.ScriptLine"]]( __result__, lambda x: typing.cast(typing.List["stream_types.ScriptLine"], x.cast_to(types, types, stream_types, True, __runtime__)), lambda x: typing.cast(typing.List["types.ScriptLine"], x.cast_to(types, types, stream_types, False, __runtime__)), __ctx__, ) def ScoreItems(self, items: typing.List["types.ScoreItem"], baml_options: BamlCallOptions = {}, ) -> baml_py.BamlSyncStream[typing.List["stream_types.ScoredItem"], typing.List["types.ScoredItem"]]: __ctx__, __result__ = self.__options.merge_options(baml_options).create_sync_stream(function_name="ScoreItems", args={ "items": items, }) return baml_py.BamlSyncStream[typing.List["stream_types.ScoredItem"], typing.List["types.ScoredItem"]]( __result__, lambda x: typing.cast(typing.List["stream_types.ScoredItem"], x.cast_to(types, types, stream_types, True, __runtime__)), lambda x: typing.cast(typing.List["types.ScoredItem"], x.cast_to(types, types, stream_types, False, __runtime__)), __ctx__, ) class BamlHttpRequestClient: __options: DoNotUseDirectlyCallManager def __init__(self, options: DoNotUseDirectlyCallManager): self.__options = options def FilterItems(self, items: typing.List["types.FilterItem"], baml_options: BamlCallOptions = {}, ) -> baml_py.baml_py.HTTPRequest: __result__ = self.__options.merge_options(baml_options).create_http_request_sync(function_name="FilterItems", args={ "items": items, }, mode="request") return __result__ def FilterScoreItems(self, items: typing.List["types.ScoreItem"], baml_options: BamlCallOptions = {}, ) -> baml_py.baml_py.HTTPRequest: __result__ = self.__options.merge_options(baml_options).create_http_request_sync(function_name="FilterScoreItems", args={ "items": items, }, mode="request") return __result__ def GenerateOutline(self, date: str,stories: typing.List["types.StoryItem"], baml_options: BamlCallOptions = {}, ) -> baml_py.baml_py.HTTPRequest: __result__ = self.__options.merge_options(baml_options).create_http_request_sync(function_name="GenerateOutline", args={ "date": date,"stories": stories, }, mode="request") return __result__ def GenerateScriptSegment(self, date: str,story_title: str,content_chunk: str,previous_context: typing.Optional[str],segment_type: str,outline_context: typing.Optional[str] = None, baml_options: BamlCallOptions = {}, ) -> baml_py.baml_py.HTTPRequest: __result__ = self.__options.merge_options(baml_options).create_http_request_sync(function_name="GenerateScriptSegment", args={ "date": date,"story_title": story_title,"content_chunk": content_chunk,"previous_context": previous_context,"segment_type": segment_type,"outline_context": outline_context, }, mode="request") return __result__ def ScoreItems(self, items: typing.List["types.ScoreItem"], baml_options: BamlCallOptions = {}, ) -> baml_py.baml_py.HTTPRequest: __result__ = self.__options.merge_options(baml_options).create_http_request_sync(function_name="ScoreItems", args={ "items": items, }, mode="request") return __result__ class BamlHttpStreamRequestClient: __options: DoNotUseDirectlyCallManager def __init__(self, options: DoNotUseDirectlyCallManager): self.__options = options def FilterItems(self, items: typing.List["types.FilterItem"], baml_options: BamlCallOptions = {}, ) -> baml_py.baml_py.HTTPRequest: __result__ = self.__options.merge_options(baml_options).create_http_request_sync(function_name="FilterItems", args={ "items": items, }, mode="stream") return __result__ def FilterScoreItems(self, items: typing.List["types.ScoreItem"], baml_options: BamlCallOptions = {}, ) -> baml_py.baml_py.HTTPRequest: __result__ = self.__options.merge_options(baml_options).create_http_request_sync(function_name="FilterScoreItems", args={ "items": items, }, mode="stream") return __result__ def GenerateOutline(self, date: str,stories: typing.List["types.StoryItem"], baml_options: BamlCallOptions = {}, ) -> baml_py.baml_py.HTTPRequest: __result__ = self.__options.merge_options(baml_options).create_http_request_sync(function_name="GenerateOutline", args={ "date": date,"stories": stories, }, mode="stream") return __result__ def GenerateScriptSegment(self, date: str,story_title: str,content_chunk: str,previous_context: typing.Optional[str],segment_type: str,outline_context: typing.Optional[str] = None, baml_options: BamlCallOptions = {}, ) -> baml_py.baml_py.HTTPRequest: __result__ = self.__options.merge_options(baml_options).create_http_request_sync(function_name="GenerateScriptSegment", args={ "date": date,"story_title": story_title,"content_chunk": content_chunk,"previous_context": previous_context,"segment_type": segment_type,"outline_context": outline_context, }, mode="stream") return __result__ def ScoreItems(self, items: typing.List["types.ScoreItem"], baml_options: BamlCallOptions = {}, ) -> baml_py.baml_py.HTTPRequest: __result__ = self.__options.merge_options(baml_options).create_http_request_sync(function_name="ScoreItems", args={ "items": items, }, mode="stream") return __result__ b = BamlSyncClient(DoNotUseDirectlyCallManager({}))