cleveragents/cleveractors-core
0
0
Fork 0
Code Issues 16 Pull Requests 2 Actions 1 Packages Projects Releases Wiki Activity

feat(runtime): add router-facing execution API
CI / quality (push) Successful in 42s
Details
CI / lint (push) Failing after 42s
Details
CI / integration_tests (push) Successful in 52s
Details
CI / typecheck (push) Successful in 1m1s
Details
CI / security (push) Failing after 1m4s
Details
CI / unit_tests (push) Failing after 3m53s
Details
CI / coverage (push) Has been skipped
Details
CI / build (push) Successful in 38s
Details
CI / status-check (push) Failing after 4s
Details

Browse Source 
Adds the public runtime API that the CleverThis router consumes:
- validate_dict(d, platform_limits) for actor config validation
- merge_configs(*dicts) for deep-merging per §3.1
- create_executor(config_dict, credentials, limits, pricing) factory
- Executor.execute(message) returning ActorResult
- ActorResult and NodeUsage dataclasses for token aggregation

Supports all four actor shapes:
- single_llm via LLMAgent
- single_graph via PureLangGraph + AgentFactory
- single_tool via ToolAgent
- multi_actor via default-actor delegation

Refs: ADR-2024, ADR-2027
This commit is contained in:
CleverThis Engineering
2026-06-04 07:03:07 +00:00
parent 6b80be2117
commit e7a7d39189
2 changed files with 555 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/cleveractors/__init__.py
+14 -1
View File
@@ -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",
]
src/cleveractors/runtime.py
+541
View File
@@ -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