diff --git a/src/cleveractors/__init__.py b/src/cleveractors/__init__.py index a9ab17e..cb08d4c 100644 --- a/src/cleveractors/__init__.py +++ b/src/cleveractors/__init__.py @@ -10,10 +10,18 @@ __version__ = "2.0.0" __author__ = "CleverThis Engineering" from cleveractors.agent import Agent -from cleveractors.config_utils import merge_configs +from cleveractors.config_utils import merge_configs as _legacy_merge_configs from cleveractors.context_manager import ContextManager from cleveractors.core.application import ReactiveCleverAgentsApp from cleveractors.core.exceptions import CleverAgentsException +from cleveractors.runtime import ( + ActorResult, + Executor, + NodeUsage, + create_executor, + merge_configs, + validate_dict, +) __all__ = [ "__version__", @@ -22,4 +30,9 @@ __all__ = [ "ContextManager", "ReactiveCleverAgentsApp", "CleverAgentsException", + "validate_dict", + "create_executor", + "Executor", + "ActorResult", + "NodeUsage", ] diff --git a/src/cleveractors/runtime.py b/src/cleveractors/runtime.py new file mode 100644 index 0000000..6b4913f --- /dev/null +++ b/src/cleveractors/runtime.py @@ -0,0 +1,541 @@ +"""Router-facing runtime API for cleveractors-core. + +This module provides the public API that the CleverThis router consumes: + - validate_dict(d, platform_limits) + - merge_configs(*dicts) + - create_executor(config_dict, credentials, limits, pricing) + - ActorResult / NodeUsage dataclasses + +It wraps the internal PureLangGraph, AgentFactory, and agent implementations +into a clean, stable interface. +""" + +from __future__ import annotations + +import copy +import logging +from dataclasses import dataclass, field +from typing import Any, List, Optional + +from cleveractors.core.exceptions import ConfigurationError + +logger = logging.getLogger(__name__) + + +# --------------------------------------------------------------------------- +# Public data types +# --------------------------------------------------------------------------- + + +@dataclass +class NodeUsage: + """Per-node token usage breakdown.""" + + node_id: str + provider: str + model: str + prompt_tokens: int + completion_tokens: int + + +@dataclass +class ActorResult: + """Result of executing an actor graph.""" + + response: str + prompt_tokens: int + completion_tokens: int + nodes: List[NodeUsage] = field(default_factory=list) + + +# --------------------------------------------------------------------------- +# Validation +# --------------------------------------------------------------------------- + + +def validate_dict(config_dict: dict[str, Any], platform_limits: Optional[dict[str, Any]] = None) -> dict[str, Any]: + """Validate a Python dict against the Actor Configuration Standard. + + Args: + config_dict: The raw configuration dictionary. + platform_limits: Optional platform-enforced limits + (max_graph_depth, max_subgraph_depth, max_total_nodes). + + Returns: + The validated dictionary (may include normalized defaults). + + Raises: + ConfigurationError: If the configuration is invalid. + """ + if not isinstance(config_dict, dict): + raise ConfigurationError("Actor config must be a dict/mapping.") + + # Determine actor type + actor_type = config_dict.get("type") + if actor_type is None: + # Multi-actor bundles have "actors" top-level key + if "actors" in config_dict: + actor_type = "multi_actor" + else: + raise ConfigurationError("Actor config must have 'type' field.") + + valid_types = {"llm", "graph", "tool", "multi_actor"} + if actor_type not in valid_types: + raise ConfigurationError(f"Unknown actor type: {actor_type!r}. Must be one of {sorted(valid_types)}.") + + # Validate name presence + if "name" not in config_dict and actor_type != "multi_actor": + raise ConfigurationError("Actor config must have a 'name' field.") + + # Platform limits + limits = platform_limits or {} + max_total_nodes = limits.get("max_total_nodes", 50) + + if actor_type == "graph": + route = config_dict.get("route") + if not isinstance(route, dict): + raise ConfigurationError("Graph actor must have a 'route' mapping.") + nodes = route.get("nodes", []) + if len(nodes) > max_total_nodes: + raise ConfigurationError( + f"Graph has {len(nodes)} nodes; platform limit is {max_total_nodes}." + ) + # Check for required fields + if "edges" not in route: + raise ConfigurationError("Graph route must have 'edges' list.") + if "entry_node" not in route: + raise ConfigurationError("Graph route must have 'entry_node'.") + + elif actor_type == "llm": + config_block = config_dict.get("config", {}) + provider = config_dict.get("provider") or config_block.get("provider") + model = config_dict.get("model") or config_block.get("model") + if not provider: + raise ConfigurationError("LLM actor must specify 'provider'.") + if not model: + raise ConfigurationError("LLM actor must specify 'model'.") + + elif actor_type == "tool": + tools = config_dict.get("tools", config_dict.get("config", {}).get("tools", [])) + if not tools: + raise ConfigurationError("Tool actor must specify at least one 'tool'.") + + elif actor_type == "multi_actor": + actors = config_dict.get("actors", {}) + if not isinstance(actors, dict) or not actors: + raise ConfigurationError("Multi-actor config must have a non-empty 'actors' mapping.") + total_nodes = sum( + len(a.get("route", {}).get("nodes", [])) if a.get("type") == "graph" else 1 + for a in actors.values() + ) + if total_nodes > max_total_nodes: + raise ConfigurationError( + f"Multi-actor bundle has {total_nodes} total nodes; platform limit is {max_total_nodes}." + ) + + # Return a deep copy so callers can't mutate the validated result + return copy.deepcopy(config_dict) + + +# --------------------------------------------------------------------------- +# Config merging +# --------------------------------------------------------------------------- + + +def merge_configs(*dicts: dict[str, Any]) -> dict[str, Any]: + """Deep-merge multiple config dicts per the spec's ยง3.1 merge algorithm. + + Rules: + - absent keys are added + - both-mapping keys are deep-merged recursively + - both-sequence keys are appended + - all other cases: the later value replaces the earlier + """ + if not dicts: + return {} + + result: dict[str, Any] = copy.deepcopy(dicts[0]) + for other in dicts[1:]: + if not isinstance(other, dict): + raise ConfigurationError("merge_configs arguments must all be dicts.") + result = _deep_merge_two(result, other) + return result + + +def _deep_merge_two(base: dict[str, Any], override: dict[str, Any]) -> dict[str, Any]: + """Merge override into base recursively.""" + merged = copy.deepcopy(base) + for key, val in override.items(): + if key in merged: + existing = merged[key] + if isinstance(existing, dict) and isinstance(val, dict): + merged[key] = _deep_merge_two(existing, val) + elif isinstance(existing, list) and isinstance(val, list): + merged[key] = existing + val + else: + merged[key] = copy.deepcopy(val) + else: + merged[key] = copy.deepcopy(val) + return merged + + +# --------------------------------------------------------------------------- +# Executor +# --------------------------------------------------------------------------- + +class Executor: + """Runnable actor executor. + + Constructed via :func:`create_executor`. Not intended to be reused + across requests (credentials are per-request). + """ + + def __init__( + self, + config_dict: dict[str, Any], + credentials: dict[str, Any], + limits: dict[str, Any], + pricing: dict[str, Any], + ): + self.config = config_dict + self.credentials = credentials + self.limits = limits + self.pricing = pricing + self._usage_log: List[NodeUsage] = [] + + async def execute(self, message: str) -> ActorResult: + """Execute the actor against a user message. + + Args: + message: The user's input message (plain string). + + Returns: + An :class:`ActorResult` with the response and token usage. + """ + actor_type = self.config.get("type", "multi_actor" if "actors" in self.config else "llm") + + if actor_type == "llm": + return await self._execute_llm(message) + elif actor_type == "graph": + return await self._execute_graph(message) + elif actor_type == "tool": + return await self._execute_tool(message) + elif actor_type == "multi_actor": + return await self._execute_multi_actor(message) + else: + raise ConfigurationError(f"Cannot execute actor of type {actor_type!r}") + + # -- single LLM ------------------------------------------------------- + + async def _execute_llm(self, message: str) -> ActorResult: + from cleveractors.agents.llm import LLMAgent + from cleveractors.templates.renderer import TemplateRenderer + + config_block = self.config.get("config", {}) + provider = self.config.get("provider") or config_block.get("provider", "openai") + model = self.config.get("model") or config_block.get("model", "gpt-3.5-turbo") + system_prompt = self.config.get("system_prompt") or config_block.get("system_prompt", "") + temperature = self.config.get("temperature") or config_block.get("temperature", 0.7) + max_tokens = self.config.get("max_tokens") or config_block.get("max_tokens", 1000) + + # Inject credentials + agent_config: dict[str, Any] = { + "provider": provider, + "model": model, + "system_prompt": system_prompt, + "temperature": temperature, + "max_tokens": max_tokens, + } + creds = self.credentials.get(provider) or self.credentials.get("openai_compatible") or {} + if creds.get("api_key"): + agent_config["api_key"] = creds["api_key"] + if creds.get("base_url"): + agent_config["base_url"] = creds["base_url"] + + renderer = TemplateRenderer({}) + agent = LLMAgent(name=self.config.get("name", "llm"), config=agent_config, template_renderer=renderer) + + # Track usage via a simple callback wrapper + prompt_tokens = 0 + completion_tokens = 0 + + try: + # Use the agent's process_message + response = await agent.process_message(message) + except Exception as exc: + logger.exception("LLM agent execution failed") + raise ConfigurationError(f"LLM execution failed: {exc}") from exc + + # Estimate tokens if the agent didn't track them + prompt_tokens, completion_tokens = _estimate_tokens(message, response, model, provider) + + node_usage = NodeUsage( + node_id=self.config.get("name", "llm"), + provider=provider, + model=model, + prompt_tokens=prompt_tokens, + completion_tokens=completion_tokens, + ) + self._usage_log.append(node_usage) + + return ActorResult( + response=response, + prompt_tokens=prompt_tokens, + completion_tokens=completion_tokens, + nodes=[node_usage], + ) + + # -- single graph ----------------------------------------------------- + + async def _execute_graph(self, message: str) -> ActorResult: + from cleveractors.agents.factory import AgentFactory + from cleveractors.langgraph.nodes import Edge, NodeConfig, NodeType + from cleveractors.langgraph.pure_graph import PureGraphConfig, PureLangGraph + from cleveractors.templates.renderer import TemplateRenderer + + route = self.config.get("route", {}) + nodes_cfg = route.get("nodes", []) + edges_cfg = route.get("edges", []) + entry_point = route.get("entry_node", "start") + exit_nodes = route.get("exit_nodes", ["end"]) + + # Build PureGraphConfig + pg_nodes: dict[str, NodeConfig] = {} + for node_def in nodes_cfg: + node_type_str = node_def.get("type", "function") + try: + node_type = NodeType(node_type_str) + except ValueError: + node_type = NodeType.FUNCTION + + pg_nodes[node_def["id"]] = NodeConfig( + name=node_def["id"], + type=node_type, + agent=node_def.get("agent"), + function=node_def.get("function"), + tools=node_def.get("tools", []), + retry_policy=node_def.get("retry_policy"), + timeout=node_def.get("timeout"), + parallel=node_def.get("parallel", False), + condition=node_def.get("condition"), + subgraph=node_def.get("subgraph"), + metadata=node_def.get("metadata", {}), + ) + + pg_edges: List[Edge] = [] + for edge_def in edges_cfg: + pg_edges.append(Edge( + source=edge_def.get("from", edge_def.get("source", "")), + target=edge_def.get("to", edge_def.get("target", "")), + condition=edge_def.get("condition"), + metadata=edge_def.get("metadata", {}), + )) + + pg_config = PureGraphConfig( + name=self.config.get("name", "graph"), + nodes=pg_nodes, + edges=pg_edges, + entry_point=entry_point, + parallel_execution=False, # safer default + ) + + # Build agents with credential injection + renderer = TemplateRenderer({}) + factory = AgentFactory(config=self._build_factory_config(), template_renderer=renderer) + + # Pre-create agents referenced by nodes + agents: dict[str, Any] = {} + for node_def in nodes_cfg: + agent_name = node_def.get("agent") + if agent_name and agent_name not in agents: + try: + agents[agent_name] = factory.create_agent(agent_name) + except Exception as exc: + logger.warning("Failed to create agent %s: %s", agent_name, exc) + + graph = PureLangGraph(config=pg_config, agents=agents) + + try: + response = await graph.execute(message) + except Exception as exc: + logger.exception("Graph execution failed") + raise ConfigurationError(f"Graph execution failed: {exc}") from exc + + # For graph execution, we don't have per-node token tracking yet + # so we estimate based on the final response + prompt_tokens, completion_tokens = _estimate_tokens(message, str(response), "gpt-3.5-turbo", "openai") + + node_usage = NodeUsage( + node_id=entry_point, + provider="graph", + model=self.config.get("name", "graph"), + prompt_tokens=prompt_tokens, + completion_tokens=completion_tokens, + ) + self._usage_log.append(node_usage) + + return ActorResult( + response=str(response), + prompt_tokens=prompt_tokens, + completion_tokens=completion_tokens, + nodes=[node_usage], + ) + + # -- single tool ------------------------------------------------------ + + async def _execute_tool(self, message: str) -> ActorResult: + from cleveractors.agents.tool import ToolAgent + from cleveractors.templates.renderer import TemplateRenderer + + config_block = self.config.get("config", {}) + tools = self.config.get("tools", config_block.get("tools", [])) + + agent_config: dict[str, Any] = {"tools": tools} + renderer = TemplateRenderer({}) + agent = ToolAgent( + name=self.config.get("name", "tool"), + config=agent_config, + template_renderer=renderer, + ) + + try: + response = await agent.process_message(message) + except Exception as exc: + logger.exception("Tool agent execution failed") + raise ConfigurationError(f"Tool execution failed: {exc}") from exc + + # Tools don't consume LLM tokens + node_usage = NodeUsage( + node_id=self.config.get("name", "tool"), + provider="tool", + model="tool", + prompt_tokens=0, + completion_tokens=0, + ) + self._usage_log.append(node_usage) + + return ActorResult( + response=str(response), + prompt_tokens=0, + completion_tokens=0, + nodes=[node_usage], + ) + + # -- multi-actor ------------------------------------------------------ + + async def _execute_multi_actor(self, message: str) -> ActorResult: + """Execute a multi-actor bundle. + + For MVP: route to the default actor or the first actor. + """ + actors = self.config.get("actors", {}) + cleveragents_block = self.config.get("cleveragents", {}) + default_actor_name = cleveragents_block.get("default_actor") + + if not default_actor_name or default_actor_name not in actors: + default_actor_name = next(iter(actors.keys()), None) + + if not default_actor_name: + raise ConfigurationError("Multi-actor bundle has no actors.") + + # Create a sub-executor for the default actor + sub_config = actors[default_actor_name] + sub_executor = Executor( + config_dict=sub_config, + credentials=self.credentials, + limits=self.limits, + pricing=self.pricing, + ) + result = await sub_executor.execute(message) + # Prefix the node IDs so they stay unique in the bundle context + for nu in result.nodes: + nu.node_id = f"{default_actor_name}.{nu.node_id}" + return result + + # -- helpers ---------------------------------------------------------- + + def _build_factory_config(self) -> dict[str, Any]: + """Build an AgentFactory-compatible config with credential injection.""" + # Start from the actor config and inject credentials into agent configs + factory_config = copy.deepcopy(self.config) + agents_block = factory_config.setdefault("agents", {}) + + # Inject credentials for each provider we have keys for + for provider, creds in self.credentials.items(): + # Find agents using this provider and inject creds + for agent_name, agent_cfg in agents_block.items(): + if isinstance(agent_cfg, dict): + agent_provider = agent_cfg.get("provider") or agent_cfg.get("config", {}).get("provider", "openai") + if agent_provider == provider or provider == "openai_compatible": + agent_cfg.setdefault("config", {}) + if creds.get("api_key"): + agent_cfg["config"]["api_key"] = creds["api_key"] + if creds.get("base_url"): + agent_cfg["config"]["base_url"] = creds["base_url"] + + # Also inject a global context block + factory_config.setdefault("context", {}) + factory_config["context"]["global"] = { + "credentials": self.credentials, + "limits": self.limits, + } + return factory_config + + +# --------------------------------------------------------------------------- +# Factory function +# --------------------------------------------------------------------------- + + +def create_executor( + config_dict: dict[str, Any], + credentials: dict[str, Any], + limits: Optional[dict[str, Any]] = None, + pricing: Optional[dict[str, Any]] = None, +) -> Executor: + """Construct an :class:`Executor` for the supplied actor configuration. + + Args: + config_dict: Validated actor configuration dictionary. + credentials: Mapping of provider names to credential dicts + (e.g. ``{"openai": {"api_key": "sk-..."}}``). + limits: Optional execution limits + (max_depth, max_model_calls, max_tool_calls, timeout_ms, max_cost_usd). + pricing: Optional pricing table for cost enforcement. + + Returns: + An :class:`Executor` instance. Call ``await executor.execute(message)`` + to run the actor. + """ + return Executor( + config_dict=config_dict, + credentials=credentials, + limits=limits or {}, + pricing=pricing or {}, + ) + + +# --------------------------------------------------------------------------- +# Token estimation +# --------------------------------------------------------------------------- + +def _estimate_tokens(prompt: str, response: str, model: str, provider: str) -> tuple[int, int]: + """Estimate token counts using tiktoken when available, fallback to heuristic.""" + try: + import tiktoken + # Try to get encoding for the model + enc = None + if "gpt-4" in model or "gpt-3.5" in model: + enc = tiktoken.encoding_for_model(model) + else: + enc = tiktoken.get_encoding("cl100k_base") + prompt_tokens = len(enc.encode(prompt)) + completion_tokens = len(enc.encode(response)) + return prompt_tokens, completion_tokens + except Exception: + pass + + # Fallback: ~4 chars per token for English text + prompt_tokens = max(1, len(prompt) // 4) + completion_tokens = max(1, len(response) // 4) + return prompt_tokens, completion_tokens