| from operator import itemgetter |
| from typing import Any, Dict, Iterable, Tuple |
| import tiktoken |
| from langchain_core.runnables import RunnablePassthrough |
|
|
|
|
| def num_tokens_from_string(string: str, encoding_name: str = "cl100k_base") -> int: |
| encoding = tiktoken.get_encoding(encoding_name) |
| num_tokens = len(encoding.encode(string)) |
| return num_tokens |
|
|
|
|
| def pass_values(x): |
| if not isinstance(x, list): |
| x = [x] |
| return {k: itemgetter(k) for k in x} |
|
|
|
|
| def prepare_chain(chain,name): |
| chain = propagate_inputs(chain) |
| chain = rename_chain(chain,name) |
| return chain |
|
|
|
|
| def propagate_inputs(chain): |
| chain_with_values = { |
| "outputs": chain, |
| "inputs": RunnablePassthrough() |
| } | RunnablePassthrough() | flatten_dict |
| return chain_with_values |
|
|
| def rename_chain(chain,name): |
| return chain.with_config({"run_name":name}) |
|
|
|
|
| |
| def _flatten_dict( |
| nested_dict: Dict[str, Any], parent_key: str = "", sep: str = "_" |
| ) -> Iterable[Tuple[str, Any]]: |
| """ |
| Generator that yields flattened items from a nested dictionary for a flat dict. |
| |
| Parameters: |
| nested_dict (dict): The nested dictionary to flatten. |
| parent_key (str): The prefix to prepend to the keys of the flattened dict. |
| sep (str): The separator to use between the parent key and the key of the |
| flattened dictionary. |
| |
| Yields: |
| (str, any): A key-value pair from the flattened dictionary. |
| """ |
| for key, value in nested_dict.items(): |
| new_key = key |
| if isinstance(value, dict): |
| yield from _flatten_dict(value, new_key, sep) |
| else: |
| yield new_key, value |
|
|
|
|
| def flatten_dict( |
| nested_dict: Dict[str, Any], parent_key: str = "", sep: str = "_" |
| ) -> Dict[str, Any]: |
| """Flattens a nested dictionary into a flat dictionary. |
| |
| Parameters: |
| nested_dict (dict): The nested dictionary to flatten. |
| parent_key (str): The prefix to prepend to the keys of the flattened dict. |
| sep (str): The separator to use between the parent key and the key of the |
| flattened dictionary. |
| |
| Returns: |
| (dict): A flat dictionary. |
| |
| """ |
| flat_dict = {k: v for k, v in _flatten_dict(nested_dict, parent_key, sep)} |
| return flat_dict |
|
|
|
|
|
|
| async def log_event(info,name,config): |
| """Helper function that will run a dummy chain with the given info |
| The astream_event function will catch this chain and stream the dict info to the logger |
| """ |
| |
| chain = RunnablePassthrough().with_config(run_name=name) |
| _ = await chain.ainvoke(info,config) |
