File size: 5,006 Bytes
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 |
from typing import Any, Optional, List
from datetime import datetime, timezone
from pydantic import BaseModel, Field
# ---------- CloudEvent ----------
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_iso() -> 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_iso(),
data=data,
)
# ---------- Permissive ancillary types you referenced ----------
class FunctionState(BaseModel):
IsFunctionCall: bool = False
IsFunctionCallResponse: bool = False
IsFunctionCallError: bool = False
IsFunctionCallStatus: bool = False
IsFunctionStillRunning: bool = False
# Convenience to set the 5-tuple like your C# SetFunctionState
def set_tuple(self, call: bool, resp: bool, err: bool, status: bool, running: bool):
self.IsFunctionCall = call
self.IsFunctionCallResponse = resp
self.IsFunctionCallError = err
self.IsFunctionCallStatus = status
self.IsFunctionStillRunning = running
class FunctionCallData(BaseModel):
# Add fields as you need; permissive placeholder
__root__: dict = Field(default_factory=dict)
class UserInfo(BaseModel):
# Add fields as you need; permissive placeholder
__root__: dict = Field(default_factory=dict)
# ---------- 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 (serialize as arrays; C# Stack<T> will read 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 (maps to your FunctionState object)
# Your C# stores this in a private field + setters; we expose the same booleans as a nested object.
IsFunctionCall: bool = False
IsFunctionCallResponse: bool = False
IsFunctionCallError: bool = False
IsFunctionCallStatus: bool = False
IsFunctionStillRunning: bool = False
# Convenience helpers (optional)
def push_llm(self, llm_name: str, new_call_id: str, new_func_name: str, new_message_id: str, new_is_processed: bool):
if self.SourceLlm:
self.LlmStack.append(self.SourceLlm)
self.SourceLlm = self.DestinationLlm
self.DestinationLlm = llm_name
if self.MessageID:
self.MessageIDStack.append(self.MessageID)
self.MessageID = new_message_id
if self.FunctionCallId:
self.FunctionCallIdStack.append(self.FunctionCallId)
self.FunctionCallId = new_call_id
if self.FunctionName:
self.FunctionNameStack.append(self.FunctionName)
self.FunctionName = new_func_name
self.IsProcessedStack.append(self.IsProcessed)
self.IsProcessed = new_is_processed
def pop_llm(self):
if self.LlmStack:
self.SourceLlm = self.LlmStack.pop()
self.DestinationLlm = self.SourceLlm
if self.MessageIDStack:
self.MessageID = self.MessageIDStack.pop()
if self.FunctionCallIdStack:
self.FunctionCallId = self.FunctionCallIdStack.pop()
if self.FunctionNameStack:
self.FunctionName = self.FunctionNameStack.pop()
if self.IsProcessedStack:
self.IsProcessed = self.IsProcessedStack.pop()
# ---------- ResultObj ----------
class ResultObj(BaseModel):
Message: str = ""
Success: bool = False
Data: Optional[Any] = None
|