from__future__importannotationsimporttracefromtypingimportTYPE_CHECKING,Any,DictifTYPE_CHECKING:# to prevent circular importfromopto.trace.nodesimportNodefromopto.trace.bundleimportbundleimportcopy@bundle()defclone(x:Any):"""This is a clone operator of x."""returncopy.deepcopy(x)
[docs]defidentity(x:Any):# identity(x) behaves the same as x.clone()returnx.clone()
# Unary operators and functions@bundle()defpos(x:Any):"""This is a pos operator of x."""return+x@bundle()defneg(x:Any):"""This is a neg operator of x."""return-x@bundle()defabs(x:Any):"""This is an abs operator of x."""returnabs(x)@bundle()definvert(x:Any):"""This is an invert operator of x."""return~x@bundle()defround(x:Any,n:Any):"""This is a round operator of x."""returnround(x,n)@bundle()deffloor(x:Any):"""This is a floor operator of x."""importmathreturnmath.floor(x)@bundle()defceil(x:Any):"""This is a ceil operator of x."""importmathreturnmath.ceil(x)@bundle()deftrunc(x:Any):"""This is a trunc operator of x."""importmathreturnmath.trunc(x)# Normal arithmetic operators@bundle()defadd(x:Any,y:Any):"""This is an add operator of x and y."""returnx+y@bundle()defsubtract(x:Any,y:Any):"""This is a subtract operator of x and y."""returnx-y@bundle()defmultiply(x:Any,y:Any):"""This is a multiply operator of x and y."""returnx*y@bundle()deffloor_divide(x:Any,y:Any):"""This is a floor_divide operator of x and y."""returnx//y@bundle()defdivide(x:Any,y:Any):"""This is a divide operator of x and y."""returnx/y@bundle()defmod(x:Any,y:Any):"""This is a mod operator of x and y."""returnx%y@bundle()defnode_divmod(x:Any,y:Any):"""This is a divmod operator of x and y."""returndivmod(x,y)@bundle()defpower(x:Any,y:Any):"""This is a power operator of x and y."""returnx**y@bundle()deflshift(x:Any,y:Any):"""This is a lshift operator of x and y."""returnx<<y@bundle()defrshift(x:Any,y:Any):"""This is a rshift operator of x and y."""returnx>>y@bundle()defand_(x:Any,y:Any):"""This is an and operator of x and y."""returnx&y@bundle()defor_(x:Any,y:Any):"""This is an or operator of x and y."""returnx|y@bundle()defxor(x:Any,y:Any):"""This is a xor operator of x and y."""returnx^y# Comparison methods@bundle()deflt(x:Any,y:Any):"""This is a lt operator of x and y."""returnx<y@bundle()defle(x:Any,y:Any):"""This is a le operator of x and y."""returnx<=y@bundle()defeq(x:Any,y:Any):"""This is an eq operator of x and y."""returnx==y@bundle()defneq(x:Any,y:Any):"""This is a not eq operator of x and y."""returnx!=y@bundle()defne(x:Any,y:Any):"""This is a ne operator of x and y."""returnx!=y@bundle()defge(x:Any,y:Any):"""This is a ge operator of x and y."""returnx>=y@bundle()defgt(x:Any,y:Any):"""This is a gt operator of x and y."""returnx>y# logical operators@bundle()defcond(condition:Any,x:Any,y:Any):"""This selects x if condition is True, otherwise y."""x,y,condition=x,y,condition# This makes sure all data are readreturnxifconditionelsey@bundle()defnot_(x:Any):"""This is a not operator of x."""returnnotx@bundle()defis_(x:Any,y:Any):"""Whether x is equal to y."""returnxisy@bundle()defis_not(x:Any,y:Any):"""Whether x is not equal to y."""returnxisnoty@bundle()defin_(x:Any,y:Any):"""Whether x is in y."""returnxiny@bundle()defnot_in(x:Any,y:Any):"""Whether x is not in y."""returnxnotiny# Indexing and slicing@bundle()defgetitem(x:Any,index:Any):"""This is a getitem operator of x based on index."""returnx[index]@bundle()defpop(x:Any,index:Any):"""This is a pop operator of x based on index."""returnx.pop(index)@bundle()deflen_(x:Any):"""This is a len operator of x."""returnlen(x)# String operators@bundle()deford_(x:Any):"""The unicode number of a character."""returnord(x)@bundle()defchr_(x:Any):"""The character of a unicode number."""returnchr(x)@bundle()defconcat(x:Any,y:Any):"""This is a concatenation operator of x and y."""returnx+y@bundle()deflower(x:Any):"""This makes all characters in x lower case."""returnx.lower()@bundle()defupper(x:Any):"""This makes all characters in x upper case."""returnx.upper()@bundle()deftitle(x:Any):"""This makes the first character to upper case and the rest to lower case."""returnx.title()@bundle()defswapcase(x:Any):"""Swaps the case of all characters: uppercase character to lowercase and vice-versa."""returnx.swapcase()@bundle()defcapitalize(x:Any):"""Converts the first character of a string to uppercase."""returnx.capitalize()@bundle()defsplit(x:Any,y:Any,maxsplit:Any=-1):"""Splits the string by finding a substring y in string x, return the first part and second part of string x without y."""returnx.split(y,maxsplit)@bundle()defstrip(x:Any,chars=None):"""Removes the leading and trailing characters of x."""returnx.strip(chars)@bundle()defreplace(x:Any,old:Any,new:Any,count:Any=-1):"""Replaces all occurrences of substring y in string x with z."""returnx.replace(old,new,count)@bundle()defformat(x:Any,*args,**kwargs):"""Fills in a string template with content, str.format()."""returnx.format(*args,**kwargs)@bundle()defjoin(x:Any,*y:Any):"""Joins a sequence y with different strs with x: "\n".join(["a", "b", "c"]) -> "a\nb\nc"."""returnx.join(y)@bundle()defnode_getattr(obj:Node,attr:str):"""This operator gets attr of obj."""returngetattr(obj,attr)@bundle(_process_inputs=False,allow_external_dependencies=True,)defcall(fun:Node,*args,**kwargs):"""This operator calls the function `fun` with args (args_0, args_1, etc.) and kwargs. If there are no args or kwargs, i.e. call(fun=function_name), the function takes no input."""# Run the function as it isfun=fun._data# Call the node with the input argumentsassertcallable(fun),"The function must be callable."output=fun(*args,**kwargs)returnoutput@bundle()defto_list(x:Any):"""This converts x to a list."""returnlist(x)@bundle()defmake_list(*args):"""This creates a list from the arguments."""returnlist(args)@bundle()defto_dict(x:Any):"""This converts x to a dictionary."""returndict(x)@bundle()defmake_dict(**kwargs):"""This creates a dictionary from the keyword arguments."""returnkwargs@bundle()defto_set(x:Any):"""This converts x to a set."""returnset(x)@bundle()defmake_set(*args):"""This creates a set from the arguments."""returnset(args)@bundle()defto_tuple(x:Any):"""This converts x to a tuple."""returntuple(x)@bundle()defmake_tuple(*args):"""This creates a tuple from the arguments."""returntuple(args)# dict operators@bundle()defkeys(x:Dict):"""Return the keys of a dictionary x as a list."""ifnotisinstance(x,dict):raiseAttributeError(f"{type(x)} object has no attribute 'values'.")return[kforkinx.keys()]@bundle()defvalues(x:Dict):"""Return the values of a dictionary x as a list."""ifnotisinstance(x,dict):raiseAttributeError(f"{type(x)} object has no attribute 'values'.")return[kforkinx.values()]# dict in-place operators@bundle()defdict_update(x:Dict,y:Dict):"""Update the dictionary x with the dictionary y."""x=copy.copy(x)x.update(y)returnx@bundle()defdict_pop(x:Dict,key:Any):"""Pop the key from the dictionary x."""x=copy.copy(x)x.pop(key)returnx@bundle()defdict_popitem(x:Dict):"""Pop the last item from the dictionary x."""x=copy.copy(x)x.popitem()returnx# list in-place operators@bundle()deflist_append(x:Any,y:Any):"""Append y to x."""x=copy.copy(x)x.append(y)returnx@bundle()deflist_clear(x:Any):"""Clear x."""x=copy.copy(x)x.clear()returnx@bundle()deflist_extend(x:Any,y:Any):"""Extend x with y."""x=copy.copy(x)x.extend(y)returnx@bundle()deflist_insert(x:Any,index:Any,y:Any):"""Insert y at index in x."""x=copy.copy(x)x.insert(index,y)returnx@bundle()deflist_pop(x:Any,index:Any):"""Pop the index from x."""x=copy.copy(x)x.pop(index)returnx@bundle()deflist_remove(x:Any,y:Any):"""Remove y from x."""x=copy.copy(x)x.remove(y)returnx@bundle()deflist_reverse(x:Any):"""Reverse x."""x=copy.copy(x)x.reverse()returnx@bundle()deflist_sort(x:Any,key:Any=None,reverse:Any=False):"""Sort x."""x=copy.copy(x)x.sort(key=key,reverse=reverse)returnx# set in-place operators@bundle()defset_add(x:Any,y:Any):"""Add y to x."""x=copy.copy(x)x.add(y)returnx@bundle()defset_clear(x:Any):"""Clear x."""x=copy.copy(x)x.clear()returnx@bundle()defset_discard(x:Any,y:Any):"""Discard y from x."""x=copy.copy(x)x.discard(y)returnx@bundle()defset_intersection_update(x:Any,y:Any):"""Update x with the intersection of x and y."""x=copy.copy(x)x.intersection_update(y)returnx@bundle()defset_pop(x:Any):"""Pop an element from x."""x=copy.copy(x)x.pop()returnx@bundle()defset_remove(x:Any,y:Any):"""Remove y from x."""x=copy.copy(x)x.remove(y)returnx@bundle()defset_symmetric_difference_update(x:Any,y:Any):"""Update x with the symmetric difference of x and y."""x=copy.copy(x)x.symmetric_difference_update(y)returnx@bundle()defset_update(x:Any,y:Any):"""Update x with y."""x=copy.copy(x)x.update(y)returnx@bundle()defcall_llm(system_prompt,*user_prompts,**kwargs):"""Query the language model of system_prompt with user_prompts."""messages=[{"role":"system","content":system_prompt}]foruser_promptinuser_prompts:messages.append({"role":"user","content":user_prompt})fromopto.utils.llmimportAutoGenLLMllm=AutoGenLLM()response=llm(messages=messages,**kwargs)returnresponse.choices[0].message.content