Framework Capabilities Guide¶
This guide covers the framework-level capabilities promoted from vertical-specific implementations to reusable components available across all verticals.
Overview¶
Victor's framework capabilities provide common patterns and functionality that can be shared across verticals. These capabilities were previously duplicated in individual verticals and have been promoted to the framework level to:
- Reduce code duplication
- Ensure consistent behavior across verticals
- Make verticals easier to maintain and extend
- Provide a unified API for common operations
External vertical authors should consume these capabilities from contract-first verticals
that inherit from victor_contracts.VerticalBase. Core runtime shims remain supported for
compatibility, but new packages should not be authored against victor.core.verticals.
Available Capabilities¶
1. Stage Builder Capability (victor.framework.capabilities.stages)¶
Provides a 7-stage generic workflow template that verticals can use and customize.
Features¶
- Generic 7-Stage Workflow: INITIAL → PLANNING → READING → ANALYSIS → EXECUTION → VERIFICATION → COMPLETION
- Stage Validation: Ensures required stages exist and transitions are valid
- Stage-to-Tool Mappings: Recommends tools for each stage
- Prompt Hints: Provides hints for each stage to include in system prompts
Usage¶
from victor.framework.capabilities import StageBuilderCapability, StageBuilderPresets
# Use default stages
capability = StageBuilderCapability()
stages = capability.get_stages()
# Get stages for a specific vertical
coding_stages = StageBuilderPresets.coding()
stages = coding_stages.get_stages()
# Get prompt hints for all stages
hints = capability.get_prompt_hints()
# Get recommended tools for a stage
tools = capability.get_tools_for_stage("EXECUTION")
# Check if a transition is valid
is_valid = capability.is_valid_transition("PLANNING", "EXECUTION")
Presets¶
Pre-configured stage builders are available for common verticals:
StageBuilderPresets.coding()- Optimized for coding verticalStageBuilderPresets.devops()- Optimized for DevOps verticalStageBuilderPresets.research()- Optimized for research verticalStageBuilderPresets.data_analysis()- Optimized for data analysis vertical
Integration in Verticals¶
from victor_contracts import VerticalBase
from victor.framework.capabilities import StageBuilderPresets
class MyVertical(VerticalBase):
def get_stages(self):
capability = StageBuilderPresets.coding()
return capability.get_stages()
2. Grounding Rules Capability (victor.framework.capabilities.grounding_rules)¶
Centralized grounding rules for common vertical constraints that ensure consistent, safe behavior.
Features¶
- Base Rules: Apply to all verticals (ground on tool output, acknowledge uncertainty)
- File Safety: Read before write, verify paths, preserve structure
- Git Safety: No force push to main, check branch status, review diffs
- Test Requirements: Verify after changes, run affected tests
- Privacy: No sensitive data exposure, redact secrets
- Tool Usage: Use appropriate tools, respect limits
Usage¶
from victor.framework.capabilities import (
GroundingRulesCapability,
GroundingRulesPresets,
RuleCategory,
)
# Get all grounding rules
capability = GroundingRulesCapability()
rules = capability.get_rules()
# Get rules for a specific category
file_rules = capability.get_rules(category=RuleCategory.FILE_SAFETY)
# Get rules for a specific vertical
coding_rules = capability.get_vertical_rules("coding")
# Get formatted text for prompts
rules_text = capability.get_rules_text(vertical="coding")
# Add custom rule
from victor.framework.capabilities import GroundingRule
custom_rule = GroundingRule(
rule_id="custom_rule",
category=RuleCategory.FILE_SAFETY,
text="Always backup files before modifying",
priority=90,
)
capability.add_rule(custom_rule)
Presets¶
Pre-configured grounding rules for common verticals:
GroundingRulesPresets.coding()- Coding-specific rulesGroundingRulesPresets.devops()- DevOps-specific rulesGroundingRulesPresets.research()- Research-specific rulesGroundingRulesPresets.data_analysis()- Data analysis-specific rulesGroundingRulesPresets.rag()- RAG-specific rules
Integration in Verticals¶
from victor_contracts import VerticalBase
from victor.framework.capabilities import GroundingRulesPresets
class MyVertical(VerticalBase):
def get_system_prompt(self):
capability = GroundingRulesPresets.coding()
grounding_rules = capability.get_rules_text(vertical="coding")
return f"{base_prompt}\n\n{grounding_rules}"
3. Validation Capability (victor.framework.capabilities.validation)¶
Pluggable validation system with common validators for tool pipelines and orchestrator.
Features¶
- File Path Validation: Validates paths, prevents traversal, checks allowed directories
- Code Syntax Validation: Validates Python, JavaScript, TypeScript, JSON, YAML syntax
- Configuration Validation: Validates required keys, value types, custom rules
- Output Format Validation: Validates length, required/forbidden patterns
- Chain Validation: Apply multiple validators in sequence
Usage¶
from victor.framework.capabilities import ValidationCapabilityProvider
# Create validation provider
provider = ValidationCapabilityProvider()
# Validate file path
result = provider.validate("file_path", "/path/to/file.txt")
if not result.is_valid:
print(f"Error: {result.error_message}")
# Validate code syntax
result = provider.validate("code_syntax_python", "print('hello')")
# Validate configuration
config = {"key": "value"}
result = provider.validate("configuration", config)
# Chain multiple validators
results = provider.validate_chain(
["file_path", "code_syntax_python"],
value,
)
# Register custom validator
from victor.framework.capabilities import Validator, ValidationResult
class CustomValidator(Validator):
@property
def validator_id(self):
return "custom"
def validate(self, value, context=None):
# Validation logic here
return ValidationResult.success()
provider.register_validator(CustomValidator())
Built-in Validators¶
FilePathValidator- File path validationCodeSyntaxValidator- Code syntax validation for multiple languagesConfigurationValidator- Configuration dictionary validationOutputFormatValidator- Output format validation
Integration Points¶
- Tool Pipeline: Validate tool arguments before execution
- Orchestrator: Validate configuration on startup
- Verticals: Register domain-specific validators
4. Safety Rules Capability (victor.framework.capabilities.safety_rules)¶
Reusable safety pattern definitions for dangerous operations across git, file, shell, docker, and network operations.
Features¶
- Pattern Matching: Regex-based pattern matching for dangerous commands
- Action Types: ALLOW, WARN, BLOCK, REQUIRE_CONFIRMATION
- Severity Levels: 1-10 severity rating for operations
- Category-Based Rules: Organized by operation type (git, file, shell, docker, network)
Usage¶
from victor.framework.capabilities import SafetyRulesCapabilityProvider, SafetyRulesPresets
# Create safety rules provider
provider = SafetyRulesCapabilityProvider()
# Check if an operation is safe
is_safe, rules = provider.check_operation("git", ["push", "--force"])
if not is_safe:
print("Operation blocked!")
# Get warnings for an operation
warnings = provider.get_warnings("git", ["reset", "--hard"])
for warning in warnings:
print(f"Warning: {warning}")
# Get required confirmations
confirmations = provider.get_required_confirmations("shell", ["mkfs", "/dev/sda"])
if confirmations:
print(f"Confirmation required: {confirmations[0]}")
# Add custom safety rule
from victor.framework.capabilities import SafetyRule, SafetyCategory, SafetyAction
custom_rule = SafetyRule(
rule_id="custom_rule",
category=SafetyCategory.SHELL,
pattern=r"dangerous_command",
description="Dangerous operation",
action=SafetyAction.REQUIRE_CONFIRMATION,
severity=9,
)
provider.add_rule(custom_rule)
Safety Categories¶
- GIT: Git operation safety (force push, branch deletion, etc.)
- FILE: File operation safety (recursive delete, overwrite, etc.)
- SHELL: Shell command safety (format, kill, system config, etc.)
- DOCKER: Docker operation safety (container removal, pruning, etc.)
- NETWORK: Network operation safety (port scanning, downloads, etc.)
- SYSTEM: System-level operations (kubectl, etc.)
Presets¶
SafetyRulesPresets.coding()- Standard coding safety rulesSafetyRulesPresets.devops()- DevOps safety rules (includes kubectl)SafetyRulesPresets.research()- Research safety rulesSafetyRulesPresets.local_mode()- Less strict for local developmentSafetyRulesPresets.production_mode()- Very strict for production
5. Task Type Hints Capability (victor.framework.capabilities.task_hints)¶
Centralized task type hints for common task types with tool budget recommendations and priority tools.
Features¶
- Task Type Detection: Match tasks based on keywords
- Tool Budget Recommendations: Recommended tool call budgets per task type
- Priority Tools: Tools to prioritize for each task type
- Prompt Hints: Hint text for inclusion in system prompts
Usage¶
from victor.framework.capabilities import TaskTypeHintCapabilityProvider, TaskTypeHintPresets
# Create task hints provider
provider = TaskTypeHintCapabilityProvider()
# Get hint for specific task type
hint = provider.get_hint("edit")
print(f"Hint: {hint.hint}")
print(f"Tool budget: {hint.tool_budget}")
print(f"Priority tools: {hint.priority_tools}")
# Get hint based on keywords
hint = provider.get_hint_for_keywords(["fix", "bug", "error"])
# Get tool budget for task
budget = provider.get_tool_budget("debug")
# Get priority tools for task
tools = provider.get_priority_tools("refactor")
# Get all prompt text formatted
prompt_text = provider.get_all_prompt_text()
Task Types¶
general- General purpose taskssearch- Searching for informationcreate- Creating new files/contentedit- Editing existing filesdebug- Finding and fixing bugsrefactor- Improving code structuretest- Writing and running testsanalyze- Deep analysisdeploy- Deployment operationsdocument- Writing documentation
Presets¶
TaskTypeHintPresets.coding()- Coding task hints (includes "review")TaskTypeHintPresets.devops()- DevOps task hints (includes "provision", "monitor")TaskTypeHintPresets.research()- Research task hints (includes "investigate", "synthesize")TaskTypeHintPresets.data_analysis()- Data analysis hints (includes "explore", "visualize")TaskTypeHintPresets.rag()- RAG task hints (includes "query", "answer")
6. Source Verification Capability (victor.framework.capabilities.source_verification)¶
Source verification for citations and references in research and RAG contexts.
Features¶
- Citation Validation: Validate citation formats
- Reference Checking: Check that references exist
- Source Reliability Scoring: Score sources by reliability
- Web Search Integration: Verify web search results
Usage¶
from victor.framework.capabilities import SourceVerificationCapabilityProvider
# Create source verification provider
provider = SourceVerificationCapabilityProvider()
# Verify a citation
result = provider.verify_citation("https://example.com", "Title")
if not result.is_valid:
print(f"Citation error: {result.error_message}")
# Check reference
result = provider.check_reference(source_id="abc123")
# Score source reliability
score = provider.score_reliability("https://trusted-source.com")
# Verify web search result
result = provider.verify_web_search_result({
"url": "https://example.com",
"title": "Example",
"snippet": "Snippet"
})
Performance Features¶
HTTP Connection Pool (victor.tools.http_pool)¶
Connection pooling for HTTP-based tools to reduce request latency by 20-30%.
Usage¶
from victor.tools.http_pool import get_http_pool, ConnectionPoolConfig
# Get connection pool
config = ConnectionPoolConfig(
max_connections=100,
max_per_host=10,
timeout_seconds=30,
)
pool = await get_http_pool(config)
# Make pooled request
response = await pool.request("GET", "https://api.example.com/data")
Configuration¶
Enable via settings:
VICTOR_HTTP_CONNECTION_POOL_ENABLED=true
VICTOR_HTTP_POOL_MAX_CONNECTIONS=100
VICTOR_HTTP_POOL_MAX_PER_HOST=10
Preload Manager (victor.framework.preload)¶
Async preload manager for warm cache initialization to reduce first-request latency by 50-70%.
Usage¶
from victor.framework.preload import PreloadManager, PreloadPriority
# Create preload manager
manager = PreloadManager(enable_parallel=True)
# Add preload tasks
manager.add_task("configuration", preload_configuration, priority=PreloadPriority.HIGH)
manager.add_task("tool_results", preload_tool_results, priority=PreloadPriority.MEDIUM)
# Execute all preloads
stats = await manager.preload_all()
print(f"Preloaded {stats.completed_tasks} tasks in {stats.total_duration_ms}ms")
Preload Tasks¶
Common preload tasks include:
- Configuration loading
- Tool results cache warming
- Embedding computation
- Model initialization
Generic Result Cache (victor.storage.cache.generic_result_cache)¶
Extended caching for non-tool results with dependency tracking and invalidation.
Usage¶
from victor.storage.cache import GenericResultCache
# Create cache
cache = GenericResultCache(max_size=1000, ttl_seconds=3600)
# Cache result
cache.put("key", {"result": "value"}, dependencies=["/path/to/file.py"])
# Get cached result
result = cache.get("key")
# Invalidate by dependency
cache.invalidate_by_dependency("/path/to/file.py")
# Get cache stats
stats = cache.get_stats()
Observability¶
Observability Manager (victor.framework.observability)¶
Unified observability manager aggregating metrics from all components.
Usage¶
from victor.framework.observability import ObservabilityManager, ObservabilityConfig
# Create manager
config = ObservabilityConfig(
max_history_size=100,
collect_system_metrics=True,
)
manager = ObservabilityManager.get_instance(config=config)
# Register a metrics source
manager.register_source("my_capability", lambda: {
"counter": 10,
"gauge": 5.5,
})
# Collect metrics from all sources
metrics = manager.collect_metrics()
# Get dashboard data
dashboard_data = manager.get_dashboard_data()
# Get historical data
historical = manager.get_historical_data(limit=100)
Dashboard CLI¶
# Launch observability dashboard
victor observability dashboard
# Show metrics for specific source
victor observability metrics my_capability
# Show historical data
victor observability history --limit 50
Best Practices¶
1. Use Presets When Available¶
Presets provide optimized configurations for common verticals:
# Good: Use preset
capability = GroundingRulesPresets.coding()
# Avoid: Manually configure unless necessary
capability = GroundingRulesCapability(custom_rules=[...])
2. Extend Rather Than Replace¶
Add custom rules/hints rather than replacing defaults:
# Good: Add custom rule to preset
capability = GroundingRulesPresets.coding()
capability.add_rule(custom_rule)
# Avoid: Replace entire capability
capability = GroundingRulesCapability(custom_rules=[...]) # Loses defaults
3. Register Custom Validators¶
Add domain-specific validators to the validation provider:
4. Use Safety Rules Consistently¶
Apply safety rules across all tools that handle dangerous operations:
safety_provider = SafetyRulesCapabilityProvider()
is_safe, rules = safety_provider.check_operation(tool_name, args)
if not is_safe:
# Handle blocked operation
5. Collect Metrics¶
Make your capabilities observable by implementing metrics collection:
from victor.framework.observability import ObservabilityManager
manager = ObservabilityManager.get_instance()
manager.register_source("my_capability", self.get_metrics)
Migration Guide¶
For Existing Verticals¶
To migrate an existing vertical to use framework capabilities:
- Replace local stage definitions with
StageBuilderPresets - Replace local grounding rules with
GroundingRulesPresets - Replace local safety rules with
SafetyRulesPresets - Replace local task hints with
TaskTypeHintPresets - Add validation using
ValidationCapabilityProvider
Example migration:
from victor_contracts import VerticalBase
from victor.framework.capabilities import StageBuilderPresets
# Before
class MyVertical(VerticalBase):
def get_stages(self):
return {
"INITIAL": {...},
"PLANNING": {...},
# ... duplicated stage definitions
}
# After
class MyVertical(VerticalBase):
def get_stages(self):
return StageBuilderPresets.coding().get_stages()
For New Verticals¶
When creating a new vertical:
- Start with the most relevant preset
- Add only the customizations specific to your vertical
- Use validation for any domain-specific constraints
- Make your vertical observable with metrics
Configuration¶
Enable Performance Features¶
# config/settings.yaml
performance:
generic_result_cache_enabled: true
http_connection_pool_enabled: true
framework_preload_enabled: true
framework_preload_parallel: true
Observability Settings¶
# config/settings.yaml
observability:
enabled: true
max_history_size: 100
collect_system_metrics: true
collection_interval_seconds: 60
API Reference¶
See the API Reference for detailed API documentation.
Troubleshooting¶
Capability Not Found¶
If you get an import error:
# Make sure capabilities are exported
from victor.framework.capabilities import (
StageBuilderCapability,
GroundingRulesCapability,
# ...
)
Custom Rules Not Applied¶
Make sure you're calling add_rule() on the capability instance, not the class:
# Correct
capability = GroundingRulesCapability()
capability.add_rule(custom_rule)
# Incorrect
GroundingRulesCapability.add_rule(custom_rule) # Class method doesn't exist
Metrics Not Collected¶
Ensure you've registered your metrics source: