File size: 4,403 Bytes
4cb71cc 2c8368f bbfbcdd 2c8368f 4cb71cc 2c8368f bbfbcdd 2c8368f bbfbcdd 2c8368f 4cb71cc 2c8368f 4cb71cc 2c8368f 4cb71cc bbfbcdd 4cb71cc 2c8368f 46c2685 2c8368f 46c2685 2c8368f 46c2685 2c8368f 46c2685 2c8368f 46c2685 2c8368f c476a18 2c8368f |
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 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 |
# models.py
from typing import Any, Optional, List, Dict
from datetime import datetime, timezone
from pydantic import BaseModel, Field, ConfigDict
# ---------- CloudEvent (Pydantic v2) ----------
class CloudEvent(BaseModel):
specversion: str = "1.0"
id: str
type: str
source: str
time: datetime
datacontenttype: str = "application/json"
data: Optional[Any] = None
@staticmethod
def now_utc() -> datetime:
return datetime.now(timezone.utc)
@classmethod
def wrap(cls, *, event_id: str, event_type: str, source: str, data: Any) -> "CloudEvent":
return cls(
id=event_id,
type=event_type or ("NullOrEmpty" if data is None else type(data).__name__),
source=source,
time=cls.now_utc(),
data=data,
)
# ---------- FunctionCallData (from your C#) ----------
class FunctionCallData(BaseModel):
function: Optional[str] = None
name: Optional[str] = None
arguments: Optional[Dict[str, Any]] = None
parameters: Optional[Dict[str, Any]] = None
# ---------- FunctionState (from your C#) ----------
class FunctionState(BaseModel):
IsFunctionCall: bool = False
IsFunctionCallResponse: bool = False
IsFunctionCallError: bool = False
IsFunctionCallStatus: bool = False
IsFunctionStillRunning: bool = False
# ---------- UserInfo (from your C#; omitting JsonIgnored/NotMapped collections) ----------
class UserInfo(BaseModel):
UserID: Optional[str] = None
DateCreated: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
HostLimit: int = 0
DisableEmail: bool = False
Status: Optional[str] = ""
Name: Optional[str] = ""
Given_name: Optional[str] = ""
Family_name: Optional[str] = ""
Nickname: Optional[str] = ""
Sub: Optional[str] = ""
Enabled: bool = True
MonitorAlertEnabled: bool = True
PredictAlertEnabled: bool = False
AccountType: Optional[str] = "Default"
Email: Optional[str] = ""
Email_verified: bool = False
Picture: Optional[str] = ""
Updated_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
LastLoginDate: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
CustomerId: Optional[str] = ""
CancelAt: Optional[datetime] = None
TokensUsed: int = 0
# NotMapped in C#: model as loose dict so it round-trips safely
LoadServer: Dict[str, Any] = Field(default_factory=dict)
# JsonIgnore MonitorIPs in C#: we do not include it here
class LLMServiceObj(BaseModel):
# allow using field names or aliases
model_config = ConfigDict(populate_by_name=True)
# strings
SessionId: str = ""
JsonFunction: str = ""
LlmMessage: str = ""
ResultMessage: str = ""
UserInput: str = ""
RequestSessionId: str = ""
FunctionName: str = ""
TimeZone: str = ""
LLMRunnerType: str = "TurboLLM"
SourceLlm: str = ""
DestinationLlm: str = ""
MessageID: str = ""
LlmSessionStartName: str = ""
SwapFunctionName: str = ""
ChatAgentLocation: str = ""
ToolsDefinitionId: Optional[str] = None
JsonToolsBuilderSpec: Optional[str] = None
# ints / bools
TokensUsed: int = 0
IsUserLoggedIn: bool = False
IsFuncAck: bool = False
IsProcessed: bool = False
IsSystemLlm: bool = False
Timeout: Optional[int] = None
# complex (rename attrs; keep JSON names via alias)
FunctionCallId: str = ""
function_call_data: FunctionCallData = Field(default_factory=FunctionCallData, alias="FunctionCallData")
user_info: UserInfo = Field(default_factory=UserInfo, alias="UserInfo")
StartTimeUTC: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
# stacks
LlmStack: List[str] = Field(default_factory=list)
FunctionCallIdStack: List[str] = Field(default_factory=list)
FunctionNameStack: List[str] = Field(default_factory=list)
IsProcessedStack: List[bool] = Field(default_factory=list)
MessageIDStack: List[str] = Field(default_factory=list)
# function state flags
IsFunctionCall: bool = False
IsFunctionCallResponse: bool = False
IsFunctionCallError: bool = False
IsFunctionCallStatus: bool = False
IsFunctionStillRunning: bool = False
class ResultObj(BaseModel):
Message: str = ""
Success: bool = False
Data: Optional[Any] = None
|