File size: 4,453 Bytes
4cb71cc 2c8368f bbfbcdd 2c8368f 4cb71cc 2c8368f bbfbcdd 2c8368f bbfbcdd 2c8368f 4cb71cc 2c8368f 4cb71cc 2c8368f 4cb71cc bbfbcdd 4cb71cc 2c8368f bbfbcdd 2c8368f 4cb71cc 2c8368f 4cb71cc 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 |
# 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
# ---------- LLMServiceObj (field names match C# exactly) ----------
class LLMServiceObj(BaseModel):
# 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
FunctionCallId: str = ""
FunctionCallData: FunctionCallData = Field(default_factory=FunctionCallData)
UserInfo: UserInfo = Field(default_factory=UserInfo)
StartTimeUTC: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
# stacks (arrays; C# Stack<T> will deserialize fine)
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 (C# stores in FunctionState; we expose flat flags like your class does)
IsFunctionCall: bool = False
IsFunctionCallResponse: bool = False
IsFunctionCallError: bool = False
IsFunctionCallStatus: bool = False
IsFunctionStillRunning: bool = False
# ---------- ResultObj (from your C#) ----------
class ResultObj(BaseModel):
Message: str = ""
Success: bool = False
Data: Optional[Any] = None
|