"""Utility & helper functions.""" from __future__ import annotations import typing from langchain.chat_models import init_chat_model from langchain_core.language_models import BaseChatModel from langchain_core.messages import BaseMessage def get_message_text(msg: BaseMessage) -> str: """Get the text content of a message.""" content = msg.content if isinstance(content, str): return content elif isinstance(content, dict): return content.get("text", "") else: txts = [c if isinstance(c, str) else (c.get("text") or "") for c in content] return "".join(txts).strip() def load_chat_model(fully_specified_name: str) -> BaseChatModel: """Load a chat model from a fully specified name. Args: fully_specified_name (str): String in the format 'provider/model'. """ provider, model = fully_specified_name.split("/", maxsplit=1) return init_chat_model(model, model_provider=provider) class State: """An object that can be used to store arbitrary state.""" _state: dict[str, typing.Any] def __init__(self, state: dict[str, typing.Any] | None = None): if state is None: state = {} super().__setattr__("_state", state) def __setattr__(self, key: typing.Any, value: typing.Any) -> None: self._state[key] = value def __getattr__(self, key: typing.Any) -> typing.Any: try: return self._state[key] except KeyError: message = "'{}' object has no attribute '{}'" raise AttributeError(message.format(self.__class__.__name__, key)) def __delattr__(self, key: typing.Any) -> None: del self._state[key]