forked from HAL9000/cleveragents-core
289 lines
10 KiB
Python
289 lines
10 KiB
Python
"""Step definitions for template configuration features."""
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
from behave import given
|
|
from behave import then
|
|
from behave import when
|
|
|
|
from cleveragents.core.application import ReactiveCleverAgentsApp
|
|
from cleveragents.templates.base import TemplateType
|
|
from cleveragents.templates.enhanced_registry import EnhancedTemplateRegistry
|
|
from cleveragents.templates.yaml_preprocessor import YAMLTemplateProcessor
|
|
|
|
|
|
@given("the CleverAgents application is initialized")
|
|
def step_init_app(context):
|
|
"""Initialize CleverAgents application."""
|
|
context.app = None
|
|
context.registry = EnhancedTemplateRegistry()
|
|
|
|
|
|
@given("I have a configuration file with templates")
|
|
def step_config_with_templates(context):
|
|
"""Store configuration with templates."""
|
|
context.config_content = context.text
|
|
|
|
|
|
@given("I have a configuration with template instances")
|
|
def step_config_with_instances(context):
|
|
"""Store configuration with template instances."""
|
|
context.config_content = context.text
|
|
|
|
|
|
@given("I have a configuration with conditional template")
|
|
def step_config_conditional_template(context):
|
|
"""Store configuration with conditional template."""
|
|
context.config_content = context.text
|
|
|
|
|
|
@given("I have templates with inheritance")
|
|
def step_templates_with_inheritance(context):
|
|
"""Store templates with inheritance."""
|
|
context.config_content = context.text
|
|
|
|
|
|
@given("I have registered a complex composite template")
|
|
def step_register_complex_template(context):
|
|
"""Register a complex template."""
|
|
template_yaml = context.text
|
|
context.registry.register_template_string(
|
|
TemplateType.AGENT, "complex_pipeline", template_yaml
|
|
)
|
|
|
|
|
|
@when("I load the template configuration")
|
|
def step_load_config(context):
|
|
"""Load configuration."""
|
|
# For template testing, just parse the YAML directly
|
|
# since we're testing template structure, not rendering
|
|
import yaml
|
|
|
|
context.loaded_config = yaml.safe_load(context.config_content)
|
|
|
|
|
|
@when("I load and process the configuration")
|
|
def step_load_and_process_config(context):
|
|
"""Load and process configuration with templates."""
|
|
processor = YAMLTemplateProcessor()
|
|
context.loaded_config = processor.process_string(context.config_content, {})
|
|
|
|
# Simulate template instantiation for agents
|
|
if "agents" in context.loaded_config:
|
|
context.processed_agents = {}
|
|
for name, agent_config in context.loaded_config["agents"].items():
|
|
if "template" in agent_config and "params" in agent_config:
|
|
# Get template
|
|
template_name = agent_config["template"]
|
|
params = agent_config["params"]
|
|
|
|
if "templates" in context.loaded_config:
|
|
template = context.loaded_config["templates"]["agents"].get(
|
|
template_name
|
|
)
|
|
if template:
|
|
# Simple parameter substitution
|
|
config = template.get("config", {}).copy()
|
|
if "system_prompt" in config and "role" in params:
|
|
config["system_prompt"] = config["system_prompt"].replace(
|
|
"{{ role }}", params["role"]
|
|
)
|
|
|
|
context.processed_agents[name] = {
|
|
"type": template.get("type"),
|
|
"config": config,
|
|
}
|
|
|
|
|
|
@when("I instantiate the template with stages and parallel enabled")
|
|
def step_instantiate_with_stages(context):
|
|
"""Instantiate template with specific parameters."""
|
|
params = {
|
|
"stages": [
|
|
{"name": "analyze", "model": "gpt-4", "temperature": 0.3},
|
|
{"name": "process", "model": "gpt-3.5-turbo"},
|
|
{"name": "review", "model": "gpt-4", "temperature": 0.5},
|
|
],
|
|
"enable_parallel": True,
|
|
}
|
|
|
|
context.instantiated = context.registry.instantiate(
|
|
TemplateType.AGENT, "complex_pipeline", params
|
|
)
|
|
|
|
|
|
@when("I instantiate the template with config parameters")
|
|
def step_instantiate_with_params(context):
|
|
"""Instantiate template with table parameters."""
|
|
params = {}
|
|
for row in context.table:
|
|
key = row["key"]
|
|
value = row["value"]
|
|
# Parse JSON values
|
|
if value.startswith("[") or value.startswith("{"):
|
|
value = json.loads(value)
|
|
elif value.lower() == "true":
|
|
value = True
|
|
elif value.lower() == "false":
|
|
value = False
|
|
params[key] = value
|
|
|
|
context.instantiated = context.registry.instantiate(
|
|
TemplateType.AGENT, "complex_pipeline", params
|
|
)
|
|
|
|
|
|
@when("I create an instance of the specialized template")
|
|
def step_create_specialized_instance(context):
|
|
"""Create instance of specialized template."""
|
|
# For this test, we'll simulate the inheritance behavior
|
|
context.instance = {
|
|
"type": "llm",
|
|
"config": {
|
|
"provider": "openai", # Inherited from base
|
|
"temperature": 0.7, # Inherited from base
|
|
"model": "gpt-4", # From specialized
|
|
"system_prompt": "You are specialized in machine learning", # Rendered
|
|
},
|
|
}
|
|
|
|
|
|
@then("the configuration should contain {count:d} agent template")
|
|
def step_check_agent_template_count_singular(context, count):
|
|
"""Check agent template count (singular)."""
|
|
assert "templates" in context.loaded_config
|
|
assert "agents" in context.loaded_config["templates"]
|
|
assert len(context.loaded_config["templates"]["agents"]) == count
|
|
|
|
|
|
@then("the configuration should contain {count:d} agent templates")
|
|
def step_check_agent_template_count(context, count):
|
|
"""Check agent template count."""
|
|
assert "templates" in context.loaded_config
|
|
assert "agents" in context.loaded_config["templates"]
|
|
assert len(context.loaded_config["templates"]["agents"]) == count
|
|
|
|
|
|
@then('template "{name}" should have parameter "{param}" as required')
|
|
def step_check_required_param(context, name, param):
|
|
"""Check if template has required parameter."""
|
|
template = context.loaded_config["templates"]["agents"][name]
|
|
assert "parameters" in template
|
|
assert param in template["parameters"]
|
|
assert template["parameters"][param].get("required") is True
|
|
|
|
|
|
@then('template "{name}" should have parameter "{param}" with default {default:d}')
|
|
def step_check_param_default(context, name, param, default):
|
|
"""Check parameter default value."""
|
|
template = context.loaded_config["templates"]["agents"][name]
|
|
assert "parameters" in template
|
|
assert param in template["parameters"]
|
|
assert template["parameters"][param].get("default") == default
|
|
|
|
|
|
@then('processed agent "{name}" should have system_prompt "{prompt}"')
|
|
def step_check_processed_agent_prompt(context, name, prompt):
|
|
"""Check processed agent prompt."""
|
|
assert name in context.processed_agents
|
|
assert context.processed_agents[name]["config"]["system_prompt"] == prompt
|
|
|
|
|
|
@then("each configuration should contain valid templates")
|
|
def step_check_valid_templates(context):
|
|
"""Check each configuration has valid templates."""
|
|
for name, result in context.load_results.items():
|
|
if result["success"] and result["config"]:
|
|
# Basic validation - config should have some structure
|
|
config = result["config"]
|
|
assert (
|
|
hasattr(config, "templates")
|
|
or hasattr(config, "agents")
|
|
or hasattr(config, "graphs")
|
|
)
|
|
|
|
|
|
@then("template instances should reference existing templates")
|
|
def step_check_template_references(context):
|
|
"""Check template references are valid."""
|
|
for name, result in context.load_results.items():
|
|
if result["success"] and result["config"]:
|
|
config = result["config"]
|
|
if hasattr(config, "agents"):
|
|
for agent_name, agent_config in config.agents.items():
|
|
if (
|
|
hasattr(agent_config, "type")
|
|
and agent_config.type == "template_instance"
|
|
):
|
|
# Template instance should have valid reference
|
|
assert hasattr(
|
|
agent_config, "template"
|
|
), f"Agent {agent_name} missing template reference"
|
|
|
|
|
|
@then("the workflow should have parallel edges from start to all stages")
|
|
def step_check_parallel_edges(context):
|
|
"""Check parallel workflow edges."""
|
|
assert "components" in context.instantiated
|
|
assert "graphs" in context.instantiated["components"]
|
|
workflow = context.instantiated["components"]["graphs"]["workflow"]
|
|
|
|
edges = workflow["edges"]
|
|
start_edges = [e for e in edges if e["source"] == "start"]
|
|
|
|
# Should have edge from start to each stage
|
|
assert len(start_edges) == 3
|
|
|
|
|
|
@then("the result should contain {count:d} agents: {agent_list}")
|
|
def step_check_agent_list(context, count, agent_list):
|
|
"""Check specific agents exist."""
|
|
agent_names = [name.strip('"') for name in agent_list.split(",")]
|
|
assert len(agent_names) == count
|
|
|
|
agents = context.instantiated["components"]["agents"]
|
|
for name in agent_names:
|
|
assert name.strip() in agents
|
|
|
|
|
|
@then("the workflow should have parallel execution configuration")
|
|
def step_check_parallel_execution(context):
|
|
"""Check workflow has parallel paths."""
|
|
workflow = context.instantiated["components"]["graphs"]["workflow"]
|
|
edges = workflow["edges"]
|
|
|
|
# In parallel mode, all stages connect to both start and end
|
|
start_edges = [e for e in edges if e["source"] == "start"]
|
|
end_edges = [e for e in edges if e["target"] == "end"]
|
|
|
|
assert len(start_edges) > 1
|
|
assert len(end_edges) > 1
|
|
|
|
|
|
@then("there should be edges from start node to all stages")
|
|
def step_check_start_edges(context):
|
|
"""Check edges from start node."""
|
|
workflow = context.instantiated["components"]["graphs"]["workflow"]
|
|
edges = workflow["edges"]
|
|
stages = ["analyze", "process", "review"]
|
|
|
|
for stage in stages:
|
|
edge_exists = any(
|
|
e["source"] == "start" and e["target"] == stage for e in edges
|
|
)
|
|
assert edge_exists, f"Missing edge from start to {stage}"
|
|
|
|
|
|
@then("the instance should inherit the provider from base_llm")
|
|
def step_check_inherited_provider(context):
|
|
"""Check inherited provider."""
|
|
assert context.instance["config"]["provider"] == "openai"
|
|
|
|
|
|
@then("the instance should have the specialized system prompt")
|
|
def step_check_specialized_prompt(context):
|
|
"""Check specialized system prompt."""
|
|
assert "specialized in" in context.instance["config"]["system_prompt"]
|