Spaces:
Sleeping
Sleeping
File size: 1,878 Bytes
71b4454 | 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 | from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any, Dict, Optional
from backend.llm.gateway import ModelGateway
@dataclass
class WorkflowState:
prompt: str
session_id: Optional[str] = None
project_id: Optional[str] = None
intent: str = "general"
response: str = ""
provider: Optional[str] = None
model: Optional[str] = None
metadata: Dict[str, Any] = field(default_factory=dict)
class WorkflowEngine:
"""
Application workflow boundary.
The workflow does not import provider implementations directly.
All model execution goes through ModelGateway.
"""
def __init__(
self,
gateway: Optional[ModelGateway] = None,
**gateway_kwargs: Any,
) -> None:
self.gateway = gateway or ModelGateway(**gateway_kwargs)
async def run(
self,
prompt: str,
intent: str = "general",
**kwargs: Any,
) -> WorkflowState:
if not isinstance(prompt, str) or not prompt.strip():
raise ValueError("prompt must be a non-empty string")
state = WorkflowState(
prompt=prompt,
intent=intent or "general",
)
result = await self.gateway.complete(
prompt=prompt,
intent=state.intent,
**kwargs,
)
if isinstance(result, str):
state.response = result
elif isinstance(result, dict):
state.response = str(
result.get("text")
or result.get("content")
or result.get("response")
or ""
)
state.provider = result.get("provider")
state.model = result.get("model")
state.metadata.update(result)
else:
state.response = str(result)
return state
|