Compilation Flow

Builder API -> AgentDefinition -> validate() -> export_yaml()
                 |                 |              |
               Domain      Schema + Gates + Static   YAML

from ainalyn import AgentBuilder, WorkflowBuilder, NodeBuilder
 
agent = (
    AgentBuilder("my-agent")
    .version("1.0.0")
    .description("My first agent")
    .add_workflow(
        WorkflowBuilder("main")
        .add_node(NodeBuilder("start").type("llm").build())
        .build()
    )
    .build()  # ← Returns AgentDefinition
)

@dataclass(frozen=True)
class AgentDefinition:
    name: str
    version: str
    description: str
    workflows: tuple[Workflow, ...]
 
    def __post_init__(self) -> None:
        # Domain validation runs immediately
        if not DefinitionRules.is_valid_name(self.name):
            raise InvalidFormatError(...)

from ainalyn import compile_agent
from pathlib import Path
 
result = compile_agent(agent, Path("agent.yaml"))

# api.py
def compile_agent(definition, output_path=None):
    service = _get_service()  # From infrastructure
    if output_path:
        return service.compile_to_file(definition, output_path)
    return service.compile(definition)

class SchemaValidator:
    """Outbound Adapter implementing IDefinitionSchemaValidator"""
 
    def validate_schema(self, definition: AgentDefinition) -> tuple[ValidationError, ...]:
        errors = []
 
        # Required fields
        if not definition.workflows:
            errors.append(ValidationError("E001", "At least one workflow required"))
 
        # Type validation
        for workflow in definition.workflows:
            if not workflow.nodes:
                errors.append(ValidationError("E002", f"Workflow '{workflow.name}' has no nodes"))
 
        return tuple(errors)

from ainalyn.domain.rules import DefinitionRules
 
class StaticAnalyzer:
    def analyze(self, definition: AgentDefinition) -> tuple[ValidationError, ...]:
        issues = []
        for workflow in definition.workflows:
            cycles = DefinitionRules.detect_circular_dependencies(workflow)
            if cycles:
                issues.append(ValidationError(code="CIRCULAR_DEPENDENCY", path=f"workflow:{workflow.name}", message="Cycle detected"))
 
            unreachable = DefinitionRules.get_unreachable_nodes(workflow)
            for node_name in unreachable:
                issues.append(ValidationError(code="UNREACHABLE_NODE", path=f"workflow:{workflow.name}", message=f"Unreachable node: {node_name}"))
 
        return tuple(issues)

class YamlExporter:
    """Outbound Adapter implementing IDefinitionSerializer"""
 
    def serialize(self, definition: AgentDefinition) -> str:
        data = {
            "name": definition.name,
            "version": definition.version,
            "description": definition.description,
            "workflows": [
                self._serialize_workflow(w) for w in definition.workflows
            ],
        }
        return yaml.dump(data, sort_keys=False)

@dataclass
class CompilationResult:
    validation_result: ValidationResult
    yaml_content: str | None  # None if validation failed
    output_path: Path | None  # Set if written to file
 
    @property
    def is_successful(self) -> bool:
        return self.validation_result.is_valid

result = compile_agent(agent)
 
if result.is_successful:
    print(result.yaml_content)
    # Submit to platform
else:
    for error in result.validation_result.errors:
        print(f"{error.code}: {error.message}")

# Cannot create invalid entity
AgentDefinition(name="invalid name!")  # Raises InvalidFormatError

# Valid entity, but missing required fields for platform
agent = AgentBuilder("test").build()  # Missing version
result = compile_agent(agent)
# result.validation_result.errors contains "version required"

# Example: missing task_goal or output_schema triggers gate violations
result = validate(agent)
for error in result.errors:
    print(error.code, error.message)

# Valid structure, but logical error
workflow = (
    WorkflowBuilder("main")
    .add_node(NodeBuilder("a").next_nodes("b").build())
    .add_node(NodeBuilder("b").next_nodes("a").build())  # Creates cycle!
    .build()
)
result = compile_agent(agent)
# result.validation_result.errors contains "cycle detected"

result = compile_agent(agent)
 
if not result.is_successful:
    for error in result.validation_result.errors:
        print(f"[{error.code}] {error.message}")
        print(f"  Path: {error.path}")
        print(f"  Suggestion: {error.suggestion}")