|
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) |
