Skip to main content

Overview

The Kapso class is the main entry point for the framework. It provides methods for research, learning, evolving, and deploying solutions.

Constructor

from kapso.kapso import Kapso

kapso = Kapso(
    config_path: Optional[str] = None,
    kg_index: Optional[str] = None,
)

Parameters

ParameterTypeDefaultDescription
config_pathstrsrc/config.yamlPath to configuration file
kg_indexstrNonePath to existing .index file for KG

Examples

# Basic initialization (no KG)
kapso = Kapso()

# With custom config
kapso = Kapso(config_path="my_config.yaml")

# With pre-indexed KG
kapso = Kapso(kg_index="data/indexes/my_knowledge.index")

# Full initialization
kapso = Kapso(
    config_path="src/config.yaml",
    kg_index="data/indexes/llm_finetuning.index",
)

Methods

research()

Run deep web research for a query.
def research(
    self,
    objective: str,
    *,
    mode: str = "both",  # "idea" | "implementation" | "both"
    depth: str = "deep",  # "light" | "deep"
) -> ResearchFindings

Parameters

ParameterTypeDefaultDescription
objectivestr-What to research
modestr"both"Research mode: “idea”, “implementation”, or “both”
depthstr"deep""light" or "deep"

Returns

ResearchFindings containing:
  • ideas: List of Source.Idea objects
  • implementations: List of Source.Implementation objects
  • report: Optional Source.ResearchReport (if mode includes study)

Example

# Research returns ResearchFindings
findings = kapso.research(
    "QLoRA best practices for fine-tuning",
    mode="both",
    depth="deep",
)

# Access results
for idea in findings.ideas:
    print(idea.to_string())

for impl in findings.implementations:
    print(impl.to_string())

# Use as context
solution = kapso.evolve(
    goal="Fine-tune LLaMA",
    context=[findings.to_string()],
)

# Or learn into KG
kapso.learn(*findings.ideas, *findings.implementations, wiki_dir="data/wikis")

learn()

Learn from knowledge sources.
def learn(
    self,
    *sources: Union[Source.Repo, Source.Solution, Idea, Implementation, ResearchReport],
    wiki_dir: str = "data/wikis",
    skip_merge: bool = False,
    kg_index: Optional[str] = None,
) -> PipelineResult

Parameters

ParameterTypeDefaultDescription
*sourcesSource.*-One or more source objects
wiki_dirstr"data/wikis"Wiki directory path
skip_mergeboolFalseOnly extract, don’t merge
kg_indexstrNoneIndex path for MCP server

Returns

PipelineResult with:
  • sources_processed: Number of sources processed
  • total_pages_extracted: Pages extracted
  • created: New pages created
  • edited: Pages merged/edited with existing
  • errors: List of errors

Example

result = kapso.learn(
    Source.Repo("https://github.com/huggingface/transformers"),
    wiki_dir="data/wikis",
)
print(f"Created: {result.created}, Edited: {result.edited}")

# Learn from research results
findings = kapso.research("LoRA best practices", mode=["idea", "implementation"], top_k=5)
kapso.learn(*findings.ideas, *findings.implementations, wiki_dir="data/wikis")

evolve()

Evolve a solution through experimentation.
def evolve(
    self,
    goal: str,
    eval_dir: Optional[str] = None,
    data_dir: Optional[str] = None,
    initial_repo: Optional[str] = None,
    max_iterations: int = 10,
    # Configuration
    mode: Optional[str] = None,
    coding_agent: Optional[str] = None,
) -> SolutionResult

Parameters

ParameterTypeDefaultDescription
goalstr-High-level objective (include success criteria)
eval_dirstrNonePath to user-provided evaluation files (copied to kapso_evaluation/)
data_dirstrNonePath to data files (copied to kapso_datasets/)
initial_repostrNoneStarting repository - see below for options
max_iterationsint10Maximum experiment iterations
modestrNoneConfig mode (GENERIC, MINIMAL, etc.)
coding_agentstrNoneCoding agent override

initial_repo Options

The initial_repo parameter accepts three types of values:
ValueBehavior
Local path ("/path/to/repo")Use existing local repository as starting point
GitHub URL ("https://github.com/org/repo")Clone the repository and use as starting point
None (default)Search for relevant workflow via workflow_search(goal). If found, clone workflow’s repo. If not found, create empty repo.

Returns

SolutionResult with:
  • goal: Original goal
  • code_path: Path to generated code
  • experiment_logs: List of experiment summaries
  • final_feedback: FeedbackResult with stop decision and score
  • metadata: Cost, iterations, final evaluation

Example

# Basic usage - agent builds evaluation dynamically
solution = kapso.evolve(
    goal="Build XGBoost classifier with AUC > 0.85",
    max_iterations=10,
)

# With user-provided evaluation and data
solution = kapso.evolve(
    goal="Optimize the model for inference speed",
    eval_dir="./my_evaluation/",
    data_dir="./my_data/",
    initial_repo="https://github.com/org/starter-repo",
)

# Check results
if solution.succeeded:
    print(f"Goal achieved with score: {solution.final_score}")
print(f"Code at: {solution.code_path}")

deploy()

Deploy a solution as running software.
def deploy(
    self,
    solution: SolutionResult,
    strategy: DeployStrategy = DeployStrategy.AUTO,
    env_vars: Optional[Dict[str, str]] = None,
    coding_agent: str = "claude_code",
) -> Software

Parameters

ParameterTypeDefaultDescription
solutionSolutionResult-Solution from evolve()
strategyDeployStrategyAUTODeployment strategy
env_varsDict[str, str]NoneEnvironment variables
coding_agentstr"claude_code"Agent for adaptation

Returns

Software instance with:
  • run(inputs): Execute with inputs
  • stop(): Stop and cleanup
  • start(): Restart
  • is_healthy(): Health check
  • logs(): Get logs

Example

deployed_program = kapso.deploy(
    solution,
    strategy=DeployStrategy.LOCAL,
    env_vars={"API_KEY": "..."},
)

result = deployed_program.run({"data_path": "./test.csv"})
deployed_program.stop()

index_kg()

Index knowledge data into the KG.
def index_kg(
    self,
    wiki_dir: Optional[str] = None,
    data_path: Optional[str] = None,
    save_to: str = None,
    search_type: Optional[str] = None,
    force: bool = False,
) -> str

Parameters

ParameterTypeDefaultDescription
wiki_dirstrNonePath to wiki directory
data_pathstrNonePath to JSON data file
save_tostr-Path to save .index file
search_typestrNoneOverride search backend
forceboolFalseClear existing data

Returns

Path to created .index file.

Example

# Index wiki pages
kapso.index_kg(
    wiki_dir="data/wikis_llm_finetuning",
    save_to="data/indexes/llm_finetuning.index",
    force=True,
)

# Index JSON graph
kapso.index_kg(
    data_path="data/kg_data.json",
    save_to="data/indexes/kg.index",
    search_type="kg_llm_navigation",
)

Data Classes

SolutionResult

@dataclass
class SolutionResult:
    goal: str
    code_path: str
    experiment_logs: List[str]
    final_feedback: Optional[FeedbackResult]
    metadata: Dict[str, Any]
    
    @property
    def succeeded(self) -> bool:
        """True if goal was achieved."""
        return self.final_feedback and self.final_feedback.stop
    
    @property
    def final_score(self) -> Optional[float]:
        """Final evaluation score if available."""
        return self.final_feedback.score if self.final_feedback else None

FeedbackResult

@dataclass
class FeedbackResult:
    stop: bool                      # Whether to stop iteration
    evaluation_valid: bool          # Whether evaluation is fair/correct
    feedback: str                   # Actionable feedback for next iteration
    score: Optional[float] = None   # Extracted evaluation score

Source Types

from kapso.kapso import Source
from kapso.knowledge_base.types import ResearchFindings

# Repository
Source.Repo(url: str, branch: str = "main")

# Solution (from evolve)
Source.Solution(obj: SolutionResult)

# Research outputs (from research)
Source.Idea(query: str, source: str, content: str)
Source.Implementation(query: str, source: str, content: str)
Source.ResearchReport(query: str, content: str)

# Multi-mode research results
ResearchFindings  # Returned by kapso.research()

DeployStrategy

from kapso.kapso import DeployStrategy

DeployStrategy.AUTO
DeployStrategy.LOCAL
DeployStrategy.DOCKER
DeployStrategy.MODAL
DeployStrategy.BENTOML
DeployStrategy.LANGGRAPH

Exceptions

KGIndexError

Raised when KG index is invalid or backend data is missing.
from kapso.kapso import KGIndexError

try:
    kapso = Kapso(kg_index="invalid.index")
except KGIndexError as e:
    print(f"Index error: {e}")

Complete Example

from kapso.kapso import Kapso, Source, DeployStrategy

# Initialize with KG
kapso = Kapso(kg_index="data/indexes/ml.index")

# Research
findings = kapso.research(
    "XGBoost best practices",
    mode="both",
    depth="deep",
)

# Learn from repo and research
kapso.learn(
    Source.Repo("https://github.com/dmlc/xgboost"),
    *findings.ideas,
    *findings.implementations,
    wiki_dir="data/wikis",
)

# Evolve with research context
solution = kapso.evolve(
    goal="XGBoost classifier with AUC > 0.9",
    context=[findings.to_string()],
    max_iterations=10,
)

# Check if goal was achieved
if solution.succeeded:
    print(f"Goal achieved with score: {solution.final_score}")

# Deploy
deployed_program = kapso.deploy(solution, strategy=DeployStrategy.LOCAL)
result = deployed_program.run({"data_path": "./test.csv"})
deployed_program.stop()

# Learn from experience
kapso.learn(Source.Solution(solution))