-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathcontext.py
More file actions
54 lines (40 loc) · 1.54 KB
/
context.py
File metadata and controls
54 lines (40 loc) · 1.54 KB
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
from typing import Any, Dict, List, Optional
from pydantic import BaseModel, Field
class UniversalDummy:
def __call__(self, *args, **kwargs): return self
def __getattr__(self, item): return self
def __await__(self):
async def _dummy(): return None
return _dummy().__await__()
def __bool__(self): return False
def __str__(self): return ""
def __iter__(self):
return iter([])
class SafeBaseModel(BaseModel):
def __getattr__(self, item):
return UniversalDummy()
class Config:
arbitrary_types_allowed = True
extra = "allow"
class FakeRunConfig(SafeBaseModel):
response_modalities: List[str] = ["TEXT"]
class FakePluginManager:
def __getattr__(self, item):
return UniversalDummy()
class FakeSession(SafeBaseModel):
id: str = "local-test-session"
state: dict = Field(default_factory=dict)
events: List[Any] = Field(default_factory=list)
class FakeContext(SafeBaseModel):
session: FakeSession = Field(default_factory=FakeSession)
input: Any = None
agent: Any = None
invocation_id: str = "local-test-invocation-001"
branch: str = "default"
flow_name: str = "default_flow"
plugin_manager: Any = Field(default_factory=FakePluginManager)
run_config: FakeRunConfig = Field(default_factory=FakeRunConfig)
agent_states: Dict[str, Any] = Field(default_factory=dict)
end_invocation: bool = False
def _get_events(self, **kwargs):
return []