mirror of
https://github.com/hoshikawa2/first_contas.git
synced 2026-07-09 18:24:20 +00:00
84 lines
2.3 KiB
Python
84 lines
2.3 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from threading import Lock
|
|
from typing import TYPE_CHECKING, Any, Callable
|
|
|
|
from agente_contas_tim.factory import CommandFactory
|
|
from agente_contas_tim.workflows.runtime_types import ActionResult
|
|
|
|
if TYPE_CHECKING:
|
|
from agente_contas_tim.agent.llm_gateway import LLMCapabilityGateway
|
|
from agente_contas_tim.workflows.runtime_types import WorkflowRunResponse
|
|
|
|
RuntimeState = dict[str, Any]
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class WorkflowRuntimeContext:
|
|
factory: CommandFactory
|
|
llm_gateway: "LLMCapabilityGateway | None" = None
|
|
workflow_runner: (
|
|
Callable[[str, dict[str, Any], str | None, int | None], "WorkflowRunResponse"]
|
|
| None
|
|
) = None
|
|
llm_callbacks: tuple[Any, ...] = ()
|
|
llm_metadata: dict[str, Any] = field(default_factory=dict)
|
|
|
|
|
|
ActionHandler = Callable[
|
|
[RuntimeState, dict[str, Any], WorkflowRuntimeContext],
|
|
ActionResult,
|
|
]
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ActionSpec:
|
|
name: str
|
|
handler: ActionHandler
|
|
source: str
|
|
|
|
|
|
class ActionRegistry:
|
|
def __init__(self) -> None:
|
|
self._actions: dict[str, ActionSpec] = {}
|
|
self._lock = Lock()
|
|
|
|
def register(self, name: str, handler: ActionHandler, *, source: str) -> None:
|
|
with self._lock:
|
|
existing = self._actions.get(name)
|
|
if existing is not None:
|
|
if existing.handler is handler:
|
|
return
|
|
raise ValueError(
|
|
f"Action {name!r} já registrada por {existing.source}; "
|
|
f"tentativa atual: {source}"
|
|
)
|
|
|
|
self._actions[name] = ActionSpec(
|
|
name=name,
|
|
handler=handler,
|
|
source=source,
|
|
)
|
|
|
|
def get(self, name: str) -> ActionHandler:
|
|
spec = self._actions.get(name)
|
|
if spec is None:
|
|
raise ValueError(f"Action não registrada: {name}")
|
|
return spec.handler
|
|
|
|
def list_names(self) -> list[str]:
|
|
return sorted(self._actions.keys())
|
|
|
|
|
|
DEFAULT_ACTION_REGISTRY = ActionRegistry()
|
|
|
|
|
|
def workflow_action(name: str):
|
|
def decorator(fn: ActionHandler) -> ActionHandler:
|
|
source = f"{fn.__module__}.{fn.__name__}"
|
|
DEFAULT_ACTION_REGISTRY.register(name, fn, source=source)
|
|
return fn
|
|
|
|
return decorator
|