this repo has no description
40
fork

Configure Feed

Select the types of activity you want to include in your feed.

at e4cd3e0e2cdbfd9b720547dee2a2f43d6b3cb694 93 lines 2.5 kB view raw
1from letta_client import Letta 2from typing import Optional 3 4def upsert_block(letta: Letta, label: str, value: str, **kwargs): 5 """ 6 Ensures that a block by this label exists. If the block exists, it will 7 replace content provided by kwargs with the values in this function call. 8 """ 9 # Get the list of blocks 10 blocks = letta.blocks.list(label=label) 11 12 # Check if we had any -- if not, create it 13 if len(blocks) == 0: 14 # Make the new block 15 new_block = letta.blocks.create( 16 label=label, 17 value=value, 18 **kwargs 19 ) 20 21 return new_block 22 23 if len(blocks) > 1: 24 raise Exception(f"{len(blocks)} blocks by the label '{label}' retrieved, label must identify a unique block") 25 26 else: 27 existing_block = blocks[0] 28 29 if kwargs.get('update', False): 30 # Remove 'update' from kwargs before passing to modify 31 kwargs_copy = kwargs.copy() 32 kwargs_copy.pop('update', None) 33 34 updated_block = letta.blocks.modify( 35 block_id = existing_block.id, 36 label = label, 37 value = value, 38 **kwargs_copy 39 ) 40 41 return updated_block 42 else: 43 return existing_block 44 45def upsert_agent(letta: Letta, name: str, **kwargs): 46 """ 47 Ensures that an agent by this label exists. If the agent exists, it will 48 update the agent to match kwargs. 49 """ 50 # Get the list of agents 51 agents = letta.agents.list(name=name) 52 53 # Check if we had any -- if not, create it 54 if len(agents) == 0: 55 # Make the new agent 56 new_agent = letta.agents.create( 57 name=name, 58 **kwargs 59 ) 60 61 return new_agent 62 63 if len(agents) > 1: 64 raise Exception(f"{len(agents)} agents by the label '{label}' retrieved, label must identify a unique agent") 65 66 else: 67 existing_agent = agents[0] 68 69 if kwargs.get('update', False): 70 # Remove 'update' from kwargs before passing to modify 71 kwargs_copy = kwargs.copy() 72 kwargs_copy.pop('update', None) 73 74 updated_agent = letta.agents.modify( 75 agent_id = existing_agent.id, 76 **kwargs_copy 77 ) 78 79 return updated_agent 80 else: 81 return existing_agent 82 83 84 85 86 87 88 89 90 91 92 93