Introduce core agent framework components: workflow persistence models, tool registry system, and reference binding logic.
This commit is contained in:
5
backend/agent/__init__.py
Normal file
5
backend/agent/__init__.py
Normal file
@@ -0,0 +1,5 @@
|
||||
"""Native Heimgeist tools and validated workflow execution."""
|
||||
|
||||
from .registry import NativeToolProvider, ToolDefinition, ToolExecutionContext
|
||||
|
||||
__all__ = ["NativeToolProvider", "ToolDefinition", "ToolExecutionContext"]
|
||||
67
backend/agent/bindings.py
Normal file
67
backend/agent/bindings.py
Normal file
@@ -0,0 +1,67 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from typing import Any, Dict, Iterable, List, Tuple
|
||||
|
||||
|
||||
REFERENCE_RE = re.compile(r"\{\{\s*([A-Za-z0-9_.-]+)\s*\}\}")
|
||||
|
||||
|
||||
class BindingError(ValueError):
|
||||
pass
|
||||
|
||||
|
||||
def lookup_reference(reference: str, values: Dict[str, Any]) -> Any:
|
||||
parts = [part for part in str(reference or "").split(".") if part]
|
||||
if not parts or parts[0] not in {"input", "run", "nodes"}:
|
||||
raise BindingError(f"Invalid reference root: {reference}")
|
||||
current: Any = values
|
||||
traversed: List[str] = []
|
||||
for part in parts:
|
||||
traversed.append(part)
|
||||
if isinstance(current, dict) and part in current:
|
||||
current = current[part]
|
||||
continue
|
||||
if isinstance(current, list) and part.isdigit() and int(part) < len(current):
|
||||
current = current[int(part)]
|
||||
continue
|
||||
raise BindingError(f"Missing reference: {'.'.join(traversed)}")
|
||||
return current
|
||||
|
||||
|
||||
def resolve_bindings(value: Any, values: Dict[str, Any]) -> Any:
|
||||
if isinstance(value, dict):
|
||||
if set(value) == {"$ref"}:
|
||||
return lookup_reference(str(value["$ref"]), values)
|
||||
return {key: resolve_bindings(item, values) for key, item in value.items()}
|
||||
if isinstance(value, list):
|
||||
return [resolve_bindings(item, values) for item in value]
|
||||
if not isinstance(value, str) or "{{" not in value:
|
||||
return value
|
||||
matches = list(REFERENCE_RE.finditer(value))
|
||||
if len(matches) == 1 and matches[0].span() == (0, len(value)):
|
||||
return lookup_reference(matches[0].group(1), values)
|
||||
output = value
|
||||
for match in reversed(matches):
|
||||
resolved = lookup_reference(match.group(1), values)
|
||||
replacement = "" if resolved is None else str(resolved)
|
||||
output = output[:match.start()] + replacement + output[match.end():]
|
||||
if "{{" in output or "}}" in output:
|
||||
raise BindingError("Malformed template expression")
|
||||
return output
|
||||
|
||||
|
||||
def iter_references(value: Any, path: str = "") -> Iterable[Tuple[str, str]]:
|
||||
if isinstance(value, dict):
|
||||
if set(value) == {"$ref"}:
|
||||
yield path or "$", str(value["$ref"])
|
||||
return
|
||||
for key, item in value.items():
|
||||
child = f"{path}.{key}" if path else key
|
||||
yield from iter_references(item, child)
|
||||
elif isinstance(value, list):
|
||||
for index, item in enumerate(value):
|
||||
yield from iter_references(item, f"{path}[{index}]")
|
||||
elif isinstance(value, str):
|
||||
for match in REFERENCE_RE.finditer(value):
|
||||
yield path or "$", match.group(1)
|
||||
116
backend/agent/models.py
Normal file
116
backend/agent/models.py
Normal file
@@ -0,0 +1,116 @@
|
||||
from datetime import datetime
|
||||
import uuid
|
||||
|
||||
from sqlalchemy import Boolean, Column, DateTime, ForeignKey, Integer, String, Text, UniqueConstraint
|
||||
from sqlalchemy.orm import relationship
|
||||
|
||||
from ..database import Base
|
||||
|
||||
|
||||
def _uuid() -> str:
|
||||
return str(uuid.uuid4())
|
||||
|
||||
|
||||
class WorkflowDefinition(Base):
|
||||
__tablename__ = "workflow_definitions"
|
||||
|
||||
id = Column(String(36), primary_key=True, default=_uuid)
|
||||
slug = Column(String(100), unique=True, nullable=False, index=True)
|
||||
name = Column(String(160), nullable=False)
|
||||
description = Column(Text, nullable=False, default="")
|
||||
current_revision_id = Column(String(36), nullable=True)
|
||||
built_in = Column(Boolean, nullable=False, default=False)
|
||||
enabled = Column(Boolean, nullable=False, default=True)
|
||||
routing_description = Column(Text, nullable=False, default="")
|
||||
routing_examples_json = Column(Text, nullable=False, default="[]")
|
||||
estimated_cost_class = Column(String(32), nullable=False, default="low")
|
||||
required_capabilities_json = Column(Text, nullable=False, default="[]")
|
||||
created_at = Column(DateTime, nullable=False, default=datetime.utcnow)
|
||||
updated_at = Column(DateTime, nullable=False, default=datetime.utcnow, onupdate=datetime.utcnow)
|
||||
|
||||
revisions = relationship(
|
||||
"WorkflowRevision",
|
||||
back_populates="workflow",
|
||||
foreign_keys="WorkflowRevision.workflow_id",
|
||||
cascade="all, delete-orphan",
|
||||
)
|
||||
|
||||
|
||||
class WorkflowRevision(Base):
|
||||
__tablename__ = "workflow_revisions"
|
||||
__table_args__ = (UniqueConstraint("workflow_id", "version", name="uq_workflow_revision_version"),)
|
||||
|
||||
id = Column(String(36), primary_key=True, default=_uuid)
|
||||
workflow_id = Column(String(36), ForeignKey("workflow_definitions.id"), nullable=False, index=True)
|
||||
version = Column(Integer, nullable=False)
|
||||
graph_json = Column(Text, nullable=False)
|
||||
created_at = Column(DateTime, nullable=False, default=datetime.utcnow)
|
||||
created_by = Column(String(64), nullable=False, default="user")
|
||||
trusted = Column(Boolean, nullable=False, default=False)
|
||||
checksum = Column(String(64), nullable=False, index=True)
|
||||
|
||||
workflow = relationship("WorkflowDefinition", back_populates="revisions", foreign_keys=[workflow_id])
|
||||
|
||||
|
||||
class WorkflowRun(Base):
|
||||
__tablename__ = "workflow_runs"
|
||||
|
||||
id = Column(String(36), primary_key=True, default=_uuid)
|
||||
workflow_id = Column(String(36), ForeignKey("workflow_definitions.id"), nullable=False, index=True)
|
||||
workflow_revision_id = Column(String(36), ForeignKey("workflow_revisions.id"), nullable=False)
|
||||
session_id = Column(String(64), nullable=True, index=True)
|
||||
user_message_id = Column(String(36), nullable=True)
|
||||
status = Column(String(32), nullable=False, default="queued", index=True)
|
||||
selection_mode = Column(String(24), nullable=False, default="direct")
|
||||
inputs_json = Column(Text, nullable=False, default="{}")
|
||||
outputs_json = Column(Text, nullable=True)
|
||||
error_json = Column(Text, nullable=True)
|
||||
metrics_json = Column(Text, nullable=False, default="{}")
|
||||
started_at = Column(DateTime, nullable=True)
|
||||
finished_at = Column(DateTime, nullable=True)
|
||||
cancellation_requested = Column(Boolean, nullable=False, default=False)
|
||||
|
||||
|
||||
class WorkflowNodeRun(Base):
|
||||
__tablename__ = "workflow_node_runs"
|
||||
__table_args__ = (UniqueConstraint("workflow_run_id", "node_id", "attempt", name="uq_node_run_attempt"),)
|
||||
|
||||
id = Column(String(36), primary_key=True, default=_uuid)
|
||||
workflow_run_id = Column(String(36), ForeignKey("workflow_runs.id"), nullable=False, index=True)
|
||||
node_id = Column(String(120), nullable=False)
|
||||
attempt = Column(Integer, nullable=False, default=1)
|
||||
status = Column(String(32), nullable=False, default="queued")
|
||||
resolved_inputs_json = Column(Text, nullable=True)
|
||||
outputs_json = Column(Text, nullable=True)
|
||||
error_json = Column(Text, nullable=True)
|
||||
started_at = Column(DateTime, nullable=True)
|
||||
finished_at = Column(DateTime, nullable=True)
|
||||
metrics_json = Column(Text, nullable=False, default="{}")
|
||||
|
||||
|
||||
class WorkflowEvent(Base):
|
||||
__tablename__ = "workflow_events"
|
||||
__table_args__ = (UniqueConstraint("workflow_run_id", "sequence", name="uq_workflow_event_sequence"),)
|
||||
|
||||
id = Column(String(36), primary_key=True, default=_uuid)
|
||||
workflow_run_id = Column(String(36), ForeignKey("workflow_runs.id"), nullable=False, index=True)
|
||||
workflow_id = Column(String(36), nullable=False)
|
||||
node_id = Column(String(120), nullable=True)
|
||||
sequence = Column(Integer, nullable=False)
|
||||
event_type = Column(String(64), nullable=False)
|
||||
timestamp = Column(DateTime, nullable=False, default=datetime.utcnow)
|
||||
payload_json = Column(Text, nullable=False, default="{}")
|
||||
|
||||
|
||||
class WorkflowConfirmation(Base):
|
||||
__tablename__ = "workflow_confirmations"
|
||||
|
||||
id = Column(String(36), primary_key=True, default=_uuid)
|
||||
workflow_run_id = Column(String(36), ForeignKey("workflow_runs.id"), nullable=False, index=True)
|
||||
node_id = Column(String(120), nullable=False)
|
||||
tool_name = Column(String(160), nullable=False)
|
||||
status = Column(String(24), nullable=False, default="pending")
|
||||
request_json = Column(Text, nullable=False, default="{}")
|
||||
response_json = Column(Text, nullable=True)
|
||||
created_at = Column(DateTime, nullable=False, default=datetime.utcnow)
|
||||
resolved_at = Column(DateTime, nullable=True)
|
||||
228
backend/agent/registry.py
Normal file
228
backend/agent/registry.py
Normal file
@@ -0,0 +1,228 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from dataclasses import dataclass, field
|
||||
import json
|
||||
from typing import Any, Awaitable, Callable, Dict, Iterable, Optional, Protocol
|
||||
|
||||
|
||||
class SchemaValidationError(ValueError):
|
||||
pass
|
||||
|
||||
|
||||
def _type_matches(value: Any, expected: str) -> bool:
|
||||
if expected == "null":
|
||||
return value is None
|
||||
if expected == "object":
|
||||
return isinstance(value, dict)
|
||||
if expected == "array":
|
||||
return isinstance(value, list)
|
||||
if expected == "string":
|
||||
return isinstance(value, str)
|
||||
if expected == "boolean":
|
||||
return isinstance(value, bool)
|
||||
if expected == "integer":
|
||||
return isinstance(value, int) and not isinstance(value, bool)
|
||||
if expected == "number":
|
||||
return isinstance(value, (int, float)) and not isinstance(value, bool)
|
||||
return True
|
||||
|
||||
|
||||
def validate_json_schema(value: Any, schema: Dict[str, Any], path: str = "$", *, partial: bool = False) -> None:
|
||||
if not isinstance(schema, dict):
|
||||
raise SchemaValidationError(f"{path}: schema must be an object")
|
||||
if "const" in schema and value != schema["const"]:
|
||||
raise SchemaValidationError(f"{path}: must equal {schema['const']!r}")
|
||||
if "enum" in schema and value not in schema["enum"]:
|
||||
raise SchemaValidationError(f"{path}: must be one of {schema['enum']!r}")
|
||||
for union_key in ("oneOf", "anyOf"):
|
||||
if union_key in schema:
|
||||
matches = 0
|
||||
errors = []
|
||||
for candidate in schema[union_key]:
|
||||
try:
|
||||
validate_json_schema(value, candidate, path, partial=partial)
|
||||
matches += 1
|
||||
except SchemaValidationError as exc:
|
||||
errors.append(str(exc))
|
||||
if (union_key == "oneOf" and matches != 1) or (union_key == "anyOf" and matches == 0):
|
||||
raise SchemaValidationError(f"{path}: does not match {union_key}: {'; '.join(errors[:3])}")
|
||||
return
|
||||
|
||||
expected = schema.get("type")
|
||||
if isinstance(expected, list):
|
||||
if not any(_type_matches(value, item) for item in expected):
|
||||
raise SchemaValidationError(f"{path}: expected one of {expected}, got {type(value).__name__}")
|
||||
elif expected and not _type_matches(value, expected):
|
||||
raise SchemaValidationError(f"{path}: expected {expected}, got {type(value).__name__}")
|
||||
|
||||
if isinstance(value, dict):
|
||||
properties = schema.get("properties") or {}
|
||||
if not partial:
|
||||
for required in schema.get("required") or []:
|
||||
if required not in value:
|
||||
raise SchemaValidationError(f"{path}.{required}: required value is missing")
|
||||
if schema.get("additionalProperties") is False:
|
||||
unknown = [key for key in value if key not in properties]
|
||||
if unknown:
|
||||
raise SchemaValidationError(f"{path}: unknown properties: {', '.join(unknown)}")
|
||||
for key, item in value.items():
|
||||
if key in properties:
|
||||
validate_json_schema(item, properties[key], f"{path}.{key}", partial=partial)
|
||||
minimum = schema.get("minProperties")
|
||||
if minimum is not None and len(value) < minimum:
|
||||
raise SchemaValidationError(f"{path}: expected at least {minimum} properties")
|
||||
elif isinstance(value, list):
|
||||
if "minItems" in schema and len(value) < int(schema["minItems"]):
|
||||
raise SchemaValidationError(f"{path}: expected at least {schema['minItems']} items")
|
||||
if "maxItems" in schema and len(value) > int(schema["maxItems"]):
|
||||
raise SchemaValidationError(f"{path}: expected at most {schema['maxItems']} items")
|
||||
item_schema = schema.get("items")
|
||||
if item_schema:
|
||||
for index, item in enumerate(value):
|
||||
validate_json_schema(item, item_schema, f"{path}[{index}]", partial=partial)
|
||||
elif isinstance(value, str):
|
||||
if "minLength" in schema and len(value) < int(schema["minLength"]):
|
||||
raise SchemaValidationError(f"{path}: string is too short")
|
||||
if "maxLength" in schema and len(value) > int(schema["maxLength"]):
|
||||
raise SchemaValidationError(f"{path}: string is too long")
|
||||
elif isinstance(value, (int, float)) and not isinstance(value, bool):
|
||||
if "minimum" in schema and value < schema["minimum"]:
|
||||
raise SchemaValidationError(f"{path}: must be at least {schema['minimum']}")
|
||||
if "maximum" in schema and value > schema["maximum"]:
|
||||
raise SchemaValidationError(f"{path}: must be at most {schema['maximum']}")
|
||||
|
||||
|
||||
def validate_schema_definition(schema: Dict[str, Any], path: str = "$schema") -> None:
|
||||
if not isinstance(schema, dict):
|
||||
raise SchemaValidationError(f"{path}: schema must be an object")
|
||||
expected = schema.get("type")
|
||||
allowed = {"object", "array", "string", "boolean", "integer", "number", "null"}
|
||||
if isinstance(expected, str) and expected not in allowed:
|
||||
raise SchemaValidationError(f"{path}.type: unsupported type {expected!r}")
|
||||
if isinstance(expected, list) and any(item not in allowed for item in expected):
|
||||
raise SchemaValidationError(f"{path}.type: contains an unsupported type")
|
||||
if expected == "object":
|
||||
properties = schema.get("properties", {})
|
||||
if not isinstance(properties, dict):
|
||||
raise SchemaValidationError(f"{path}.properties: must be an object")
|
||||
for key, child in properties.items():
|
||||
validate_schema_definition(child, f"{path}.properties.{key}")
|
||||
required = schema.get("required", [])
|
||||
if not isinstance(required, list) or any(not isinstance(item, str) for item in required):
|
||||
raise SchemaValidationError(f"{path}.required: must be a string array")
|
||||
missing = [item for item in required if item not in properties]
|
||||
if missing:
|
||||
raise SchemaValidationError(f"{path}.required: unknown properties: {', '.join(missing)}")
|
||||
if expected == "array" and "items" in schema:
|
||||
validate_schema_definition(schema["items"], f"{path}.items")
|
||||
for key in ("oneOf", "anyOf"):
|
||||
if key in schema:
|
||||
if not isinstance(schema[key], list) or not schema[key]:
|
||||
raise SchemaValidationError(f"{path}.{key}: must be a non-empty array")
|
||||
for index, child in enumerate(schema[key]):
|
||||
validate_schema_definition(child, f"{path}.{key}[{index}]")
|
||||
|
||||
|
||||
ToolHandler = Callable[[Dict[str, Any], "ToolExecutionContext"], Awaitable[Dict[str, Any]]]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ToolDefinition:
|
||||
name: str
|
||||
description: str
|
||||
input_schema: Dict[str, Any]
|
||||
output_schema: Dict[str, Any]
|
||||
handler: ToolHandler
|
||||
risk_level: str = "low"
|
||||
timeout_seconds: float = 120.0
|
||||
result_size_limit: int = 120_000
|
||||
requires_confirmation: bool = False
|
||||
llm_call: bool = False
|
||||
|
||||
def manifest(self) -> Dict[str, Any]:
|
||||
return {
|
||||
"name": self.name,
|
||||
"description": self.description,
|
||||
"inputSchema": self.input_schema,
|
||||
"outputSchema": self.output_schema,
|
||||
"riskLevel": self.risk_level,
|
||||
"timeoutSeconds": self.timeout_seconds,
|
||||
"resultSizeLimit": self.result_size_limit,
|
||||
"requiresConfirmation": self.requires_confirmation,
|
||||
}
|
||||
|
||||
def ollama_manifest(self) -> Dict[str, Any]:
|
||||
return {
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": self.name,
|
||||
"description": self.description,
|
||||
"parameters": self.input_schema,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
@dataclass
|
||||
class ToolExecutionContext:
|
||||
run_id: str
|
||||
workflow_id: str
|
||||
node_id: str
|
||||
session_id: Optional[str]
|
||||
selection_mode: str
|
||||
explicit_user_action: bool
|
||||
emit: Callable[[str, Dict[str, Any]], Awaitable[None]]
|
||||
cancellation_event: asyncio.Event
|
||||
db_factory: Callable[[], Any]
|
||||
registry: Optional["NativeToolProvider"] = None
|
||||
run_values: Dict[str, Any] = field(default_factory=dict)
|
||||
request_confirmation: Optional[Callable[[ToolDefinition, Dict[str, Any]], Awaitable[bool]]] = None
|
||||
consume_llm_call: Optional[Callable[[], None]] = None
|
||||
show_thinking: bool = False
|
||||
|
||||
|
||||
class ToolProvider(Protocol):
|
||||
def list_tools(self) -> Iterable[Dict[str, Any]]: ...
|
||||
def get_tool(self, name: str) -> ToolDefinition: ...
|
||||
async def call_tool(self, name: str, arguments: Dict[str, Any], context: ToolExecutionContext) -> Dict[str, Any]: ...
|
||||
|
||||
|
||||
class NativeToolProvider:
|
||||
def __init__(self) -> None:
|
||||
self._tools: Dict[str, ToolDefinition] = {}
|
||||
|
||||
def register(self, definition: ToolDefinition) -> None:
|
||||
if definition.name in self._tools:
|
||||
raise ValueError(f"Duplicate tool name: {definition.name}")
|
||||
validate_schema_definition(definition.input_schema, f"{definition.name}.inputSchema")
|
||||
validate_schema_definition(definition.output_schema, f"{definition.name}.outputSchema")
|
||||
self._tools[definition.name] = definition
|
||||
|
||||
def list_tools(self) -> Iterable[Dict[str, Any]]:
|
||||
return [tool.manifest() for tool in sorted(self._tools.values(), key=lambda item: item.name)]
|
||||
|
||||
def get_tool(self, name: str) -> ToolDefinition:
|
||||
try:
|
||||
return self._tools[name]
|
||||
except KeyError as exc:
|
||||
raise KeyError(f"Unknown tool: {name}") from exc
|
||||
|
||||
async def call_tool(self, name: str, arguments: Dict[str, Any], context: ToolExecutionContext) -> Dict[str, Any]:
|
||||
definition = self.get_tool(name)
|
||||
validate_json_schema(arguments, definition.input_schema)
|
||||
if context.cancellation_event.is_set():
|
||||
raise asyncio.CancelledError()
|
||||
if definition.llm_call and context.consume_llm_call:
|
||||
context.consume_llm_call()
|
||||
try:
|
||||
result = await asyncio.wait_for(
|
||||
definition.handler(arguments, context),
|
||||
timeout=definition.timeout_seconds,
|
||||
)
|
||||
except asyncio.TimeoutError as exc:
|
||||
raise TimeoutError(f"Tool {name} exceeded {definition.timeout_seconds:g}s") from exc
|
||||
validate_json_schema(result, definition.output_schema)
|
||||
size = len(json.dumps(result, ensure_ascii=False, default=str))
|
||||
if size > definition.result_size_limit:
|
||||
raise ValueError(f"Tool {name} result exceeded {definition.result_size_limit} characters")
|
||||
return result
|
||||
126
backend/agent/schemas.py
Normal file
126
backend/agent/schemas.py
Normal file
@@ -0,0 +1,126 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, Dict, List, Literal, Optional
|
||||
|
||||
from pydantic import BaseModel, ConfigDict, Field
|
||||
|
||||
|
||||
class WorkflowPosition(BaseModel):
|
||||
x: float = 0
|
||||
y: float = 0
|
||||
|
||||
|
||||
class WorkflowNode(BaseModel):
|
||||
model_config = ConfigDict(extra="forbid")
|
||||
|
||||
id: str
|
||||
type: Literal["input", "output", "tool", "prompt", "agent", "condition", "merge", "template", "select", "limit"]
|
||||
position: WorkflowPosition = Field(default_factory=WorkflowPosition)
|
||||
config: Dict[str, Any] = Field(default_factory=dict)
|
||||
|
||||
|
||||
class WorkflowEdge(BaseModel):
|
||||
model_config = ConfigDict(extra="forbid")
|
||||
|
||||
id: str
|
||||
source: str
|
||||
source_handle: str = "output"
|
||||
target: str
|
||||
target_handle: str = "input"
|
||||
|
||||
|
||||
class WorkflowLimits(BaseModel):
|
||||
maximum_nodes: int = 40
|
||||
maximum_tool_calls: int = 20
|
||||
maximum_llm_calls: int = 8
|
||||
maximum_runtime_seconds: int = 300
|
||||
maximum_result_chars: int = 120_000
|
||||
maximum_concurrency: int = 4
|
||||
|
||||
|
||||
class WorkflowGraph(BaseModel):
|
||||
model_config = ConfigDict(extra="forbid")
|
||||
|
||||
schema_version: int = 1
|
||||
inputs: Dict[str, Dict[str, Any]] = Field(default_factory=dict)
|
||||
outputs: Dict[str, Dict[str, Any]] = Field(default_factory=dict)
|
||||
nodes: List[WorkflowNode]
|
||||
edges: List[WorkflowEdge]
|
||||
limits: WorkflowLimits = Field(default_factory=WorkflowLimits)
|
||||
|
||||
|
||||
class ValidationIssue(BaseModel):
|
||||
code: str
|
||||
message: str
|
||||
node_id: Optional[str] = None
|
||||
field_path: Optional[str] = None
|
||||
|
||||
|
||||
class WorkflowValidationResult(BaseModel):
|
||||
valid: bool
|
||||
errors: List[ValidationIssue] = Field(default_factory=list)
|
||||
warnings: List[ValidationIssue] = Field(default_factory=list)
|
||||
|
||||
|
||||
class WorkflowCreateRequest(BaseModel):
|
||||
name: str
|
||||
slug: Optional[str] = None
|
||||
description: str = ""
|
||||
routing_description: str = ""
|
||||
routing_examples: List[str] = Field(default_factory=list)
|
||||
enabled: bool = True
|
||||
graph: Dict[str, Any]
|
||||
|
||||
|
||||
class WorkflowUpdateRequest(BaseModel):
|
||||
name: Optional[str] = None
|
||||
description: Optional[str] = None
|
||||
routing_description: Optional[str] = None
|
||||
routing_examples: Optional[List[str]] = None
|
||||
enabled: Optional[bool] = None
|
||||
|
||||
|
||||
class WorkflowRevisionRequest(BaseModel):
|
||||
graph: Dict[str, Any]
|
||||
created_by: str = "user"
|
||||
|
||||
|
||||
class WorkflowValidateRequest(BaseModel):
|
||||
graph: Dict[str, Any]
|
||||
|
||||
|
||||
class WorkflowRunRequest(BaseModel):
|
||||
session_id: Optional[str] = None
|
||||
message: str = ""
|
||||
model: str
|
||||
selection_mode: Literal["direct", "automatic", "explicit"] = "direct"
|
||||
workflow_id: Optional[str] = None
|
||||
workflow_revision_id: Optional[str] = None
|
||||
library_slug: Optional[str] = None
|
||||
web_search_enabled: bool = False
|
||||
searx_url: Optional[str] = None
|
||||
searx_engines: List[str] = Field(default_factory=list)
|
||||
attachments: List[Dict[str, Any]] = Field(default_factory=list)
|
||||
vision_model: Optional[str] = None
|
||||
transcription_model: Optional[str] = None
|
||||
stream: bool = True
|
||||
router_model: Optional[str] = None
|
||||
show_thinking: bool = False
|
||||
generation_options: Dict[str, Any] = Field(default_factory=dict)
|
||||
sample_inputs: Dict[str, Any] = Field(default_factory=dict)
|
||||
explicit_user_action: bool = False
|
||||
|
||||
|
||||
class ConfirmationResponseRequest(BaseModel):
|
||||
approved: bool
|
||||
note: Optional[str] = None
|
||||
|
||||
|
||||
class RouterSelectRequest(BaseModel):
|
||||
message: str
|
||||
recent_messages: List[Dict[str, Any]] = Field(default_factory=list)
|
||||
attachments: List[Dict[str, Any]] = Field(default_factory=list)
|
||||
library_slug: Optional[str] = None
|
||||
model: Optional[str] = None
|
||||
chat_model: Optional[str] = None
|
||||
confidence_threshold: float = 0.55
|
||||
156
backend/agent/validation.py
Normal file
156
backend/agent/validation.py
Normal file
@@ -0,0 +1,156 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from collections import defaultdict, deque
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
from pydantic import ValidationError
|
||||
|
||||
from .bindings import iter_references
|
||||
from .registry import NativeToolProvider, SchemaValidationError, validate_json_schema, validate_schema_definition
|
||||
from .schemas import ValidationIssue, WorkflowGraph, WorkflowValidationResult
|
||||
|
||||
|
||||
HOST_LIMITS = {
|
||||
"maximum_nodes": 80,
|
||||
"maximum_tool_calls": 40,
|
||||
"maximum_llm_calls": 16,
|
||||
"maximum_runtime_seconds": 600,
|
||||
"maximum_result_chars": 240_000,
|
||||
"maximum_concurrency": 8,
|
||||
}
|
||||
RUN_REFERENCE_FIELDS = {
|
||||
"session_id", "chat_model", "router_model", "vision_model", "transcription_model",
|
||||
"library_slug", "searx_url", "searx_engines", "messages", "attachments",
|
||||
"generation_options", "context_blocks", "target_message_id", "target_url",
|
||||
}
|
||||
|
||||
|
||||
def _issue(code: str, message: str, node_id: Optional[str] = None, field_path: Optional[str] = None) -> ValidationIssue:
|
||||
return ValidationIssue(code=code, message=message, node_id=node_id, field_path=field_path)
|
||||
|
||||
|
||||
def validate_workflow_graph(raw_graph: Dict[str, Any], registry: NativeToolProvider) -> WorkflowValidationResult:
|
||||
errors: List[ValidationIssue] = []
|
||||
warnings: List[ValidationIssue] = []
|
||||
try:
|
||||
graph = WorkflowGraph.model_validate(raw_graph)
|
||||
except ValidationError as exc:
|
||||
for item in exc.errors():
|
||||
errors.append(_issue("malformed_graph", item["msg"], field_path=".".join(str(p) for p in item["loc"])))
|
||||
return WorkflowValidationResult(valid=False, errors=errors)
|
||||
|
||||
if graph.schema_version != 1:
|
||||
errors.append(_issue("unsupported_schema_version", "Only workflow schema_version 1 is supported.", field_path="schema_version"))
|
||||
|
||||
for name, schema in {**graph.inputs, **graph.outputs}.items():
|
||||
try:
|
||||
validate_schema_definition(schema, f"schema.{name}")
|
||||
except SchemaValidationError as exc:
|
||||
errors.append(_issue("malformed_schema", str(exc), field_path=f"inputs.{name}"))
|
||||
|
||||
node_ids = [node.id for node in graph.nodes]
|
||||
edge_ids = [edge.id for edge in graph.edges]
|
||||
for identifier, values, path in (("node", node_ids, "nodes"), ("edge", edge_ids, "edges")):
|
||||
seen = set()
|
||||
for value in values:
|
||||
if not value:
|
||||
errors.append(_issue(f"empty_{identifier}_id", f"{identifier.title()} IDs cannot be empty.", field_path=path))
|
||||
elif value in seen:
|
||||
errors.append(_issue(f"duplicate_{identifier}_id", f"Duplicate {identifier} ID: {value}", field_path=path))
|
||||
seen.add(value)
|
||||
|
||||
nodes = {node.id: node for node in graph.nodes}
|
||||
input_nodes = [node for node in graph.nodes if node.type == "input"]
|
||||
output_nodes = [node for node in graph.nodes if node.type == "output"]
|
||||
if not input_nodes:
|
||||
errors.append(_issue("missing_input_node", "A workflow must contain an input node."))
|
||||
if not output_nodes:
|
||||
errors.append(_issue("missing_output_node", "A workflow must contain an output node."))
|
||||
|
||||
outgoing: Dict[str, List[Any]] = defaultdict(list)
|
||||
incoming: Dict[str, List[Any]] = defaultdict(list)
|
||||
for edge in graph.edges:
|
||||
if edge.source not in nodes:
|
||||
errors.append(_issue("missing_edge_source", f"Edge source does not exist: {edge.source}", field_path=f"edges.{edge.id}.source"))
|
||||
if edge.target not in nodes:
|
||||
errors.append(_issue("missing_edge_target", f"Edge target does not exist: {edge.target}", field_path=f"edges.{edge.id}.target"))
|
||||
if edge.source in nodes and edge.target in nodes:
|
||||
outgoing[edge.source].append(edge)
|
||||
incoming[edge.target].append(edge)
|
||||
|
||||
indegree = {node_id: len(incoming[node_id]) for node_id in nodes}
|
||||
queue = deque(node_id for node_id, degree in indegree.items() if degree == 0)
|
||||
visited: List[str] = []
|
||||
while queue:
|
||||
node_id = queue.popleft()
|
||||
visited.append(node_id)
|
||||
for edge in outgoing[node_id]:
|
||||
indegree[edge.target] -= 1
|
||||
if indegree[edge.target] == 0:
|
||||
queue.append(edge.target)
|
||||
if len(visited) != len(nodes):
|
||||
errors.append(_issue("cycle", "Workflow schema version 1 must be a DAG; cycles are not allowed."))
|
||||
|
||||
reachable = set()
|
||||
frontier = [node.id for node in input_nodes]
|
||||
while frontier:
|
||||
node_id = frontier.pop()
|
||||
if node_id in reachable:
|
||||
continue
|
||||
reachable.add(node_id)
|
||||
frontier.extend(edge.target for edge in outgoing[node_id])
|
||||
for node in graph.nodes:
|
||||
if node.type not in {"input"} and node.id not in reachable:
|
||||
errors.append(_issue("unreachable_node", f"Node {node.id} is not reachable from an input node.", node.id))
|
||||
|
||||
for node in graph.nodes:
|
||||
if node.type == "tool":
|
||||
tool_name = str(node.config.get("tool") or "")
|
||||
try:
|
||||
tool = registry.get_tool(tool_name)
|
||||
except KeyError:
|
||||
errors.append(_issue("unknown_tool", f"Unknown tool: {tool_name or '(missing)'}", node.id, "config.tool"))
|
||||
else:
|
||||
arguments = node.config.get("arguments", {})
|
||||
if not isinstance(arguments, dict):
|
||||
errors.append(_issue("invalid_tool_arguments", "Tool arguments must be an object.", node.id, "config.arguments"))
|
||||
else:
|
||||
literal_arguments = {
|
||||
key: value for key, value in arguments.items()
|
||||
if not (isinstance(value, dict) and set(value) == {"$ref"})
|
||||
}
|
||||
try:
|
||||
validate_json_schema(literal_arguments, tool.input_schema, partial=True)
|
||||
except SchemaValidationError as exc:
|
||||
errors.append(_issue("invalid_tool_arguments", str(exc), node.id, "config.arguments"))
|
||||
if node.type == "condition":
|
||||
operation = node.config.get("operation")
|
||||
allowed = {"equals", "not_equals", "exists", "greater_than", "less_than", "contains", "array_not_empty", "score_at_least"}
|
||||
if operation not in allowed:
|
||||
errors.append(_issue("invalid_condition", f"Unsupported condition operation: {operation}", node.id, "config.operation"))
|
||||
handles = {edge.source_handle for edge in outgoing[node.id]}
|
||||
if not {"true", "false"}.issubset(handles):
|
||||
errors.append(_issue("invalid_branch", "Condition nodes require both true and false outgoing branches.", node.id))
|
||||
if node.type == "prompt" and not node.config.get("model_source"):
|
||||
errors.append(_issue("missing_model_source", "Prompt nodes require model_source.", node.id, "config.model_source"))
|
||||
|
||||
for field_path, reference in iter_references(node.config, f"nodes.{node.id}.config"):
|
||||
parts = reference.split(".")
|
||||
if parts[0] == "input" and (len(parts) < 2 or parts[1] not in graph.inputs):
|
||||
errors.append(_issue("missing_reference", f"Unknown workflow input reference: {reference}", node.id, field_path))
|
||||
elif parts[0] == "run" and (len(parts) < 2 or parts[1] not in RUN_REFERENCE_FIELDS):
|
||||
errors.append(_issue("missing_reference", f"Unknown run reference: {reference}", node.id, field_path))
|
||||
elif parts[0] == "nodes" and (len(parts) < 3 or parts[1] not in nodes):
|
||||
errors.append(_issue("missing_reference", f"Unknown node reference: {reference}", node.id, field_path))
|
||||
elif parts[0] not in {"input", "run", "nodes"}:
|
||||
errors.append(_issue("missing_reference", f"Invalid reference: {reference}", node.id, field_path))
|
||||
|
||||
limits = graph.limits.model_dump()
|
||||
for key, maximum in HOST_LIMITS.items():
|
||||
value = limits[key]
|
||||
if value < 1 or value > maximum:
|
||||
errors.append(_issue("excessive_limit", f"{key} must be between 1 and {maximum}.", field_path=f"limits.{key}"))
|
||||
if len(graph.nodes) > graph.limits.maximum_nodes:
|
||||
errors.append(_issue("excessive_nodes", "The graph contains more nodes than its maximum_nodes limit.", field_path="nodes"))
|
||||
|
||||
return WorkflowValidationResult(valid=not errors, errors=errors, warnings=warnings)
|
||||
Reference in New Issue
Block a user