File size: 5,555 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 8d27c84 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 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 |
# 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
def set_as_call(self) -> "LLMServiceObj":
self.IsFunctionCall = True
self.IsFunctionCallResponse = False
self.IsFunctionCallError = False
self.IsFunctionCallStatus = False
self.IsFunctionStillRunning = False
return self
def set_as_call_error(self) -> "LLMServiceObj":
self.IsFunctionCall = True
self.IsFunctionCallResponse = False
self.IsFunctionCallError = True
self.IsFunctionCallStatus = False
self.IsFunctionStillRunning = False
return self
def set_as_response_complete(self) -> "LLMServiceObj":
self.IsFunctionCall = False
self.IsFunctionCallResponse = True
self.IsFunctionCallError = False
self.IsFunctionCallStatus = False
self.IsFunctionStillRunning = False
self.IsProcessed = True
return self
def set_as_not_call(self) -> "LLMServiceObj":
self.IsFunctionCall = False
self.IsFunctionCallResponse = False
self.IsFunctionCallError = False
self.IsFunctionCallStatus = False
self.IsFunctionStillRunning = False
return self
class ResultObj(BaseModel):
Message: str = ""
Success: bool = False
Data: Optional[Any] = None
|