Agent class for CopilotKit
| 16 | |
| 17 | |
| 18 | class Agent(ABC): |
| 19 | """Agent class for CopilotKit""" |
| 20 | |
| 21 | def __init__( |
| 22 | self, |
| 23 | *, |
| 24 | name: str, |
| 25 | description: Optional[str] = None, |
| 26 | ): |
| 27 | self.name = name |
| 28 | self.description = description |
| 29 | |
| 30 | if not re.match(r"^[a-zA-Z0-9_-]+$", name): |
| 31 | raise ValueError( |
| 32 | f"Invalid agent name '{name}': " |
| 33 | + "must consist of alphanumeric characters, underscores, and hyphens only" |
| 34 | ) |
| 35 | |
| 36 | @abstractmethod |
| 37 | def execute( # pylint: disable=too-many-arguments |
| 38 | self, |
| 39 | *, |
| 40 | state: dict, |
| 41 | config: Optional[dict] = None, |
| 42 | messages: List[Message], |
| 43 | thread_id: str, |
| 44 | actions: Optional[List[ActionDict]] = None, |
| 45 | meta_events: Optional[List[MetaEvent]] = None, |
| 46 | **kwargs, |
| 47 | ): |
| 48 | """Execute the agent""" |
| 49 | |
| 50 | @abstractmethod |
| 51 | async def get_state( |
| 52 | self, |
| 53 | *, |
| 54 | thread_id: str, |
| 55 | ): |
| 56 | """Default get_state implementation""" |
| 57 | return { |
| 58 | "threadId": thread_id or "", |
| 59 | "threadExists": False, |
| 60 | "state": {}, |
| 61 | "messages": [], |
| 62 | } |
| 63 | |
| 64 | def dict_repr(self) -> AgentDict: |
| 65 | """Dict representation of the action""" |
| 66 | return {"name": self.name, "description": self.description or ""} |
no outgoing calls
searching dependent graphs…