File size: 2,648 Bytes
139fefe 37b1e7a 48e003d 37b1e7a 139fefe 48e003d 139fefe 37b1e7a 139fefe 48e003d |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 |
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})
# Drawn from langchain utils and modified to remove the parent key
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) |