helix-framework 0.5.1

Production-grade AI agent framework — cost governance, memory, caching, multi-agent teams, and built-in eval

Helix

A Python framework for building production AI agents.

Helix gives you agents that actually behave in production: hard budget limits, semantic caching for repeated queries, opt-in persistent memory (SQLite, zero extra dependency), MCP tool support, agent handoffs, multi-agent teams, YAML-based task pipelines, a LangGraph-compatible StateGraph, and a 6-scorer eval suite. It works out of the box with OpenAI, Anthropic, Gemini, Groq, Mistral, and 8 other providers.

The import helix API is intentionally close to what you already know from AutoGen, CrewAI, and LangGraph, but with the production layer those frameworks leave to you: cost governance, caching, memory, observability, and safety controls.

Table of Contents

• Installation
• Quickstart
• Agents
• Class-Based Agents
• Preset Agents
• Agent Pipelines
• Tools
• MCP Tools
• Tasks and Pipelines
• YAML Configuration
• Multi-Agent Teams
• Handoffs
• Group Chat
• Workflows
• StateGraph
• Sessions
• Budget Enforcement
• Guardrails
• Event Hooks
• Evaluation
• Framework Adapters
• CLI
• Architecture
• Supported Providers
• Contributing

---

Installation

pip install helix-framework                        # core only (pydantic required)
pip install "helix-framework[gemini]"              # + Google Gemini (free tier available)
pip install "helix-framework[openai,anthropic]"    # + OpenAI and Anthropic
pip install "helix-framework[all]"                 # all providers

From source:

git clone https://github.com/sarcasticdhruv/helix-agent
cd helix-agent
pip install -e ".[all]"

API key setup

The easiest way is the persistent config store:

helix config set GOOGLE_API_KEY    "AIza..."    # Gemini, free tier works fine
helix config set OPENAI_API_KEY    "sk-..."
helix config set ANTHROPIC_API_KEY "sk-ant-..."

Keys are saved to ~/.helix/config.json. Helix picks the best available model automatically when multiple keys are set.

Or use environment variables directly:

# Linux / macOS
export GOOGLE_API_KEY="AIza..."

# Windows PowerShell
$env:GOOGLE_API_KEY = "AIza..."

---

Quickstart

import helix

agent = helix.Agent(
    name="Researcher",
    role="Research analyst",
    goal="Find accurate, cited answers.",
)

result = helix.run(agent, "What is quantum entanglement?")
print(result.output)
print(f"Cost:  ${result.cost_usd:.4f}")
print(f"Steps: {result.steps}")

For the fastest possible start, use helix.quick() — no config objects needed:

import helix

agent = helix.quick("You are a concise Python tutor.", budget_usd=0.10)
result = helix.run(agent, "Explain list comprehensions.")
print(result.output)

Inside an async function, call run_async or agent.run directly:

import asyncio
import helix

async def main():
    agent = helix.Agent(
        name="Researcher",
        role="Research analyst",
        goal="Find accurate answers.",
    )
    result = await agent.run("What is quantum entanglement?")
    print(result.output)

asyncio.run(main())

helix.quick() parameters:

Parameter	Description
system_prompt	The agent's purpose as plain instructions
name	Agent name shown in traces (default "Agent")
model	Model string, e.g. "gpt-4o". Auto-detected if omitted
tools	List of @helix.tool-decorated functions
budget_usd	Hard spend cap per run (default 0.10)
on_event	Optional async/sync event callback (see Event Hooks)

---

Agents

import helix

agent = helix.Agent(
    name="Analyst",
    role="Senior data analyst",
    goal="Analyze datasets and produce concise summaries.",

    # Optional: rich background context that shapes agent behaviour
    backstory=(
        "You have 8 years of experience in financial data analysis. "
        "You prefer bullet-point summaries over long prose."
    ),

    # Model selection with automatic fallback
    model=helix.ModelConfig(
        primary="gpt-4o",
        fallback_chain=["gpt-4o-mini", "gemini-2.0-flash"],
        temperature=0.3,
    ),

    # Hard cost limit
    budget=helix.BudgetConfig(budget_usd=1.00),
    mode=helix.AgentMode.PRODUCTION,

    # Memory
    memory=helix.MemoryConfig(short_term_limit=20),

    # Semantic caching (cost reduction on repeated/similar queries)
    cache=helix.CacheConfig(enabled=True, semantic_threshold=0.92),
)

result = helix.run(agent, "Summarize last quarter's sales trends.")

AgentResult fields: output, cost_usd, steps, model_used, cache_hits, cache_savings_usd, tool_calls, run_id, duration_s, trace.

Agents also expose LangChain-compatible aliases: agent.invoke(task) (sync) and await agent.ainvoke(task) (async), both equivalent to helix.run() / await agent.run().

---

Class-Based Agents

The @helix.agent decorator turns any class into an Agent factory. Tools become methods decorated with @helix.tool, and the class docstring becomes the system prompt.

import helix

@helix.agent(model="claude-sonnet-4-6", budget_usd=2.00)
class WebResearcher:
    """
    You are an expert web researcher.
    Find accurate, up-to-date information and always cite sources.
    """

    @helix.tool(description="Search the web for recent information.")
    async def search(self, query: str) -> list[dict]:
        from helix.tools.builtin import web_search
        return await web_search(query)

    @helix.tool(description="Fetch and read a URL.")
    async def fetch(self, url: str) -> str:
        from helix.tools.builtin import fetch_url
        result = await fetch_url(url)
        return result.get("content", "")

# The decorator returns a factory; call it to get an Agent instance
researcher = WebResearcher()
result = helix.run(researcher, "Latest AI safety research 2026")

@helix.agent options:

Parameter	Description
model	LLM model string. Auto-detected if omitted
budget_usd	Spending cap per run (default 0.50)
mode	"explore" (default) or "production"
name	Override agent name; defaults to class name
backstory	Rich background context injected into system prompt

---

Preset Agents

helix.presets provides nine ready-made agent factories so you can start in one line:

from helix.presets import web_researcher, writer, coder, summariser

# Single agent
result = helix.run(web_researcher(), "Top AI papers this week")

# Code generation
result = helix.run(coder("TypeScript"), "Write a UUID v4 generator")

# Pipe agents together with |
research_and_write = web_researcher() | writer()
result = research_and_write.run_sync("Write a report on quantum computing")

Available presets:

Factory	Description
web_researcher(budget_usd, model, max_results)	Web search + URL fetching
writer(style, budget_usd)	Polished prose from bullet points or research
summariser(style, budget_usd)	Compress long text into a summary
fact_checker(budget_usd)	Verify claims against web sources
coder(language, budget_usd, allow_file_io)	Code generation and debugging
code_reviewer(language, budget_usd)	Bug, style, and security review
data_analyst(budget_usd)	Statistical analysis with calculator tool
api_agent(base_url, auth_token, budget_usd)	REST API orchestration
assistant(domain, budget_usd)	General-purpose assistant for any domain

---

Agent Pipelines

Use the | operator to wire agents into a sequential pipeline. Each agent's output becomes the next agent's input.

from helix.presets import web_researcher, summariser, writer

# Build with | operator
pipeline = web_researcher() | summariser() | writer(style="blog post")
result = pipeline.run_sync("Quantum computing advances in 2026")
print(result.output)
print(f"Cost: ${result.cost_usd:.4f}")

Or use helix.chain() for a more explicit form:

pipeline = helix.chain(web_researcher(), summariser(), writer())
result = pipeline.run_sync("Quantum computing advances in 2026")

AgentPipeline also exposes:

• await pipeline.run(task) — async version
• pipeline.agents — list of Agent instances in the chain

---

Tools

import helix

@helix.tool(
    description="Search the web for current information.",
    timeout=15.0,
    retries=2,
)
async def web_search(query: str, max_results: int = 5) -> list:
    # your implementation here
    return [{"title": "...", "url": "...", "snippet": "..."}]


@helix.tool(description="Read a file from disk.")
async def read_file(path: str) -> str:
    with open(path) as f:
        return f.read()


agent = helix.Agent(
    name="Researcher",
    role="Research analyst",
    goal="Find answers using web search.",
    tools=[web_search, read_file],
)

result = helix.run(agent, "What are the latest AI headlines?")

Built-in tools (13 included):

import helix.tools.builtin  # registers tools globally

# web_search, fetch_url, read_file, write_file, list_directory,
# calculator, json_query, get_datetime, get_env,
# text_stats, extract_urls, sleep, execute_python

The execute_python tool runs sandboxed Python code in an isolated subprocess. Dangerous modules (subprocess, os.system, ctypes, pty, multiprocessing) are blocked. Returns {"success", "stdout", "stderr", "returncode"} with a configurable timeout (default 15 s).

from helix.tools.builtin import execute_python

# Use inside an agent that needs to run arbitrary Python
agent = helix.Agent(
    name="Calculator",
    role="Python executor",
    goal="Run Python snippets and return results.",
    tools=[execute_python],
)

Use helix.discover_tools() to list every tool registered in the global registry (built-ins + any @helix.tool functions loaded at import time).

Breaking change (v0.5): agents no longer automatically get every globally-registered tool. Pass the tools you want explicitly via tools=[...], or opt in to the old behavior with Agent(..., inherit_global_tools=True).

---

MCP Tools

Connect to any Model Context Protocol server and use its tools like any other Helix tool. Requires pip install "helix-framework[mcp]".

import helix
from helix.tools.mcp import MCPToolSource

async def main():
    async with MCPToolSource(command="npx", args=["-y", "@some/mcp-server"]) as tools:
        agent = helix.Agent(
            name="Bot",
            role="Assistant",
            goal="Use the connected MCP tools to help the user.",
            tools=tools,
        )
        result = await agent.run("...")
        print(result.output)

Without the context manager (when the connection needs to outlive one call):

source = MCPToolSource(command="python", args=["-m", "my_mcp_server"])
tools = await source.connect()
# ... use tools across multiple agent runs ...
await source.close()

---

Tasks and Pipelines

Tasks are first-class declarative units of work. They chain outputs together, support output validation with guardrails, and can write results to files. This is the Helix equivalent of CrewAI's Task + crew.kickoff().

import helix

researcher = helix.Agent(
    name="Researcher",
    role="Research analyst",
    goal="Find accurate information on {topic}.",
    backstory="You specialize in academic and technical research.",
)
writer = helix.Agent(
    name="Writer",
    role="Technical writer",
    goal="Write clear articles on {topic}.",
)

research = helix.Task(
    description="Research the latest advances in {topic}.",
    expected_output="A list of 5 key findings with sources.",
    agent=researcher,
)
article = helix.Task(
    description="Write a 3-paragraph article based on the research.",
    expected_output="A well-structured article, no jargon.",
    agent=writer,
    context=[research],        # automatically receives research output
    output_file="article.md",  # saved to disk when done
)

pipeline = helix.Pipeline(tasks=[research, article])
result = pipeline.kickoff(inputs={"topic": "quantum computing"})
print(result.final_output)
print(f"Total cost: ${result.total_cost_usd:.4f}")

Task options:

Parameter	Description
context	List of Tasks whose outputs are passed as context
output_schema	Pydantic model for structured output
guardrail	Validation function or string description
guardrails	List of validation functions (chained)
guardrail_max_retries	How many times to retry on validation failure (default 3)
output_file	Path to write the task output
async_execution	Run this task concurrently with others
callback	Called with TaskOutput after completion
markdown	Instruct the agent to format output as Markdown

Validation with guardrails:

from helix import Task, TaskOutput

def must_be_under_300_words(result: TaskOutput):
    words = len(result.raw.split())
    if words > 300:
        return False, f"Too long: {words} words (max 300)"
    return True, result.raw

task = helix.Task(
    description="Write a short summary of {topic}.",
    expected_output="A summary under 300 words.",
    agent=writer,
    guardrail=must_be_under_300_words,
    guardrail_max_retries=2,
)

You can also pass a plain string and Helix uses the agent's own LLM to validate:

task = helix.Task(
    description="Write a product description for {product}.",
    expected_output="A concise, professional product description.",
    agent=writer,
    guardrail="Must be professional, under 100 words, and avoid superlatives.",
)

Accessing task output:

result = pipeline.kickoff(inputs={"topic": "AI safety"})

for task_output in result.task_outputs:
    print(f"Task:  {task_output.summary}")
    print(f"Raw:   {task_output.raw}")
    if task_output.pydantic:
        print(f"Model: {task_output.pydantic}")

---

YAML Configuration

Define agents and tasks in YAML files for cleaner project structure:

# agents.yaml
researcher:
  role: Senior Research Analyst
  goal: Find cutting-edge developments in {topic}.
  backstory: You work at a leading tech think tank with access to academic databases.

writer:
  role: Content Strategist
  goal: Write engaging, accurate articles about {topic}.
  backstory: You have 5 years of experience writing technical content for developers.

# tasks.yaml
research_task:
  description: Research the latest developments in {topic}.
  expected_output: A structured report with at least 5 key findings.
  agent: researcher

write_task:
  description: Write a concise article based on the research.
  expected_output: A 3-paragraph article written for a developer audience.
  agent: writer
  context: [research_task]
  output_file: output/article.md

import helix

pipeline = helix.from_yaml(
    "agents.yaml",
    "tasks.yaml",
    inputs={"topic": "large language models"},
)
result = pipeline.kickoff()
print(result.final_output)

Or use the lower-level helpers:

from helix.core.yaml_config import load_agents, load_tasks, load_pipeline

agents   = load_agents("agents.yaml", inputs={"topic": "LLMs"})
tasks    = load_tasks("tasks.yaml", agents, inputs={"topic": "LLMs"})
pipeline = load_pipeline(tasks)
result   = pipeline.kickoff()

---

Multi-Agent Teams

Teams coordinate multiple agents with three execution strategies.

import helix

searcher = helix.Agent(name="Searcher", role="Web researcher",   goal="Find sources.")
analyst  = helix.Agent(name="Analyst",  role="Data analyst",     goal="Analyze data.")
writer   = helix.Agent(name="Writer",   role="Technical writer", goal="Write reports.")

# sequential: searcher output feeds into analyst, then into writer
team = helix.Team(
    name="research-team",
    agents=[searcher, analyst, writer],
    strategy="sequential",
    budget_usd=5.00,
)

result = team.run_sync("Write a report on renewable energy trends.")
print(result.final_output)
print(f"Total cost: ${result.total_cost_usd:.4f}")

Strategies:

• sequential - each agent receives the previous agent's output as its input
• parallel - all agents run on the same input concurrently, outputs returned as a list
• hierarchical - a lead agent decomposes the task and delegates subtasks to specialists

lead = helix.Agent(name="Lead", role="Project lead", goal="Decompose and delegate tasks.")

team = helix.Team(
    name="product-team",
    agents=[searcher, analyst, writer],
    strategy="hierarchical",
    lead=lead,
)

---

Handoffs

Give an agent a list of handoffs and it can transfer the conversation to a specialist mid-run — visible to the model as a normal tool call (transfer_to_<name>), not a hidden orchestration decision.

import helix

billing = helix.Agent(
    name="BillingAgent",
    role="Billing specialist",
    goal="Handle billing questions, invoices, refunds, and payment issues.",
)
triage = helix.Agent(
    name="Triage",
    role="Customer support triage agent",
    goal="Figure out what the customer needs. Transfer billing questions to the billing agent.",
    handoffs=[billing],
)

result = await triage.run("My invoice #4471 shows double charges, can you help?")
print(result.output)          # produced by BillingAgent, not Triage
print(result.handoff_chain)   # ["Triage"] — the path this run took before reaching its answer
print(result.cost_usd)        # includes cost from every agent in the chain

handoffs can be combined with guardrails and any other Agent option. Chained handoffs (A → B → C) accumulate in handoff_chain in order.

---

Group Chat

Group chat puts multiple agents in a shared multi-turn conversation. This is Helix's equivalent of AutoGen's GroupChat.

import asyncio
import helix

ceo    = helix.ConversableAgent(name="CEO",    role="CEO",    goal="Make strategic decisions.")
cto    = helix.ConversableAgent(name="CTO",    role="CTO",    goal="Assess technical risk.")
lawyer = helix.ConversableAgent(name="Lawyer", role="Lawyer", goal="Flag compliance issues.")

chat = helix.GroupChat(
    agents=[ceo, cto, lawyer],
    max_rounds=6,
    speaker_selection="round_robin",  # or "auto", "random", or a callable
    termination_keyword="AGREED",
)

async def main():
    result = await chat.run("Should we migrate our core product to microservices?")
    print(result.transcript())
    print(f"Rounds: {result.rounds}, Cost: ${result.total_cost_usd:.4f}")

asyncio.run(main())

Speaker selection:

Value	Behavior
round_robin	Agents speak in order (default)
auto	A coordinator LLM picks the most relevant next speaker
random	Random selection each round
callable	fn(agents, history) -> Agent

Termination:

chat = helix.GroupChat(
    agents=[...],
    max_rounds=10,
    termination_keyword="FINAL ANSWER",
    termination_fn=lambda msgs: len(msgs) > 8,
)

Human in the loop:

human = helix.HumanAgent(name="You")   # prompts the terminal each turn

chat = helix.GroupChat(
    agents=[agent1, agent2, human],
    max_rounds=5,
)

---

StateGraph

helix.StateGraph is a LangGraph-compatible directed graph engine for building complex agentic pipelines with cycles, conditional branching, and checkpoint persistence.

import helix
from typing import TypedDict

class State(TypedDict):
    topic: str
    draft: str
    ready: bool

researcher = helix.presets.web_researcher()
writer     = helix.presets.writer()

async def research_node(state: State) -> dict:
    result = await researcher.run(state["topic"])
    return {"draft": result.output}

async def write_node(state: State) -> dict:
    result = await writer.run(state["draft"])
    return {"draft": result.output, "ready": True}

def router(state: State) -> str:
    return helix.END if state.get("ready") else "write"

graph = (
    helix.StateGraph(State)
    .add_node("research", research_node)
    .add_node("write", write_node)
    .add_edge("research", "write")
    .add_conditional_edges("write", router, {"write": "write", helix.END: helix.END})
    .set_entry_point("research")
    .compile()
)

result = graph.run_sync({"topic": "Quantum computing in 2026", "draft": "", "ready": False})
print(result["draft"])

StateGraph API:

Method	Description
.add_node(name, fn)	Register an async or sync callable as a graph node
.add_edge(a, b)	Unconditional edge from node a to node b
.add_conditional_edges(node, fn, mapping)	Routing function determines next node
.set_entry_point(node)	First node executed
.set_finish_point(node)	Node that signals graph completion
.compile(checkpoint_dir=...)	Returns CompiledGraph

CompiledGraph execution:

result = graph.run_sync(initial_state)       # synchronous
result = await graph.run(initial_state)      # async
result = await graph.ainvoke(initial_state)  # LangChain-compatible alias
async for state in graph.stream(initial_state):  # step-by-step streaming
    print(state)

helix.END and helix.START are exported directly from the top-level helix namespace.

Pass checkpoint_dir=".helix/checkpoints" to .compile() to save state after every node, enabling resume-after-crash for long-running graphs.

---

Workflows

Workflows are step-based directed pipelines with retry, timeout, fallback, and branching.

import helix

@helix.step(name="search", retry=2, timeout_s=10.0)
async def search_step(query: str) -> list:
    return []  # your search implementation

@helix.step(name="summarise")
async def summarise_step(results: list) -> str:
    return "\n".join(str(r) for r in results)

pipeline = (
    helix.Workflow("research-pipeline")
    .then(search_step)
    .then(summarise_step)
    .with_budget(2.00)
)

result = pipeline.run_sync("quantum computing trends 2025")
print(result.final_output)

---

Sessions

Sessions give an agent persistent memory across multiple turns.

import asyncio
import helix

async def main():
    agent = helix.Agent(name="Bot", role="Assistant", goal="Help users.")
    session = helix.Session(agent=agent)
    await session.start()

    r1 = await session.send("My name is Alice.")
    r2 = await session.send("What is my name?")   # remembers: Alice
    print(r2.output)

    await session.end()

asyncio.run(main())

---

Event Hooks

Attach an on_event callback to any agent to receive live telemetry without any extra config — no trace files, no external services.

import helix
from helix.core.hooks import HookEvent

async def my_hook(event: HookEvent) -> None:
    if event.type == "tool_call":
        print(f"  → {event.data['tool_name']}({event.data['args']})")
    elif event.type == "step_end":
        print(f"  ✓ step {event.step} — ${event.cost_so_far:.4f} spent")
    elif event.type == "llm_call":
        print(f"  [LLM] {event.data['model']}")
    elif event.type == "cache_hit":
        print(f"  [CACHE] saved ${event.data['saved_usd']:.4f}")

agent = helix.Agent(
    name="Researcher",
    role="Research analyst",
    goal="Find information.",
    on_event=my_hook,  # sync or async
)

Event types:

Event	Payload keys
step_start	step
step_end	step, output_preview
llm_call	model, messages
llm_response	model, tokens, finish_reason
tool_call	tool_name, args
tool_result	tool_name, result_preview
tool_error	tool_name, error
cache_hit	similarity, saved_usd
done	output_preview, steps, cost_usd
error	error

Hook errors are silently swallowed so they never affect agent execution. Both sync and async callables are supported.

---

Budget Enforcement

import helix

agent = helix.Agent(
    name="Bot",
    role="Assistant",
    goal="Help users.",
    budget=helix.BudgetConfig(
        budget_usd=0.50,
        warn_at_pct=0.8,
        strategy=helix.BudgetStrategy.DEGRADE,  # step down to cheaper model instead of stopping
    ),
    mode=helix.AgentMode.PRODUCTION,
)

try:
    result = helix.run(agent, "Write a 10,000 word essay on climate change...")
except helix.BudgetExceededError as e:
    print(f"Budget hit: ${e.spent_usd:.4f} of ${e.budget_usd:.4f}")

With BudgetStrategy.DEGRADE, Helix steps down through the fallback chain as the budget depletes rather than stopping outright.

---

Guardrails

Guardrails run on both the incoming task and the model's output — pass built-in guardrail names to Agent(...) directly:

import helix

agent = helix.Agent(
    name="Bot",
    role="Assistant",
    goal="Help users.",
    guardrails=["prompt_injection", "pii_redactor", "length_guard"],
)

Name	Behavior
prompt_injection	Blocks common jailbreak/injection phrasings (instruction override, persona hijack, restriction bypass, "developer mode", system-prompt exfiltration). Heuristic pattern matching, not a trained classifier.
pii_redactor	Redacts email, phone, SSN, credit card, and IP address patterns. Never blocks — cleans and passes through.
length_guard	Blocks responses shorter than min_chars or longer than max_chars (defaults: 1 / 100,000).
keyword_block	Blocks content containing configured keywords (default: none — must be built directly for a custom list).
schema_guard	Validates that JSON-shaped output actually parses as JSON.

A guardrail violation raises helix.errors.GuardrailViolationError.

guardrails=[...] on Agent(...) only supports built-in names with default parameters (build_guardrail_chain under the hood). For a custom-configured guardrail — a specific keyword list, flag instead of block mode — build and call a GuardrailChain directly wherever you need it:

from helix.safety.guardrails import GuardrailChain, KeywordBlockGuard, PromptInjectionGuard

chain = GuardrailChain([
    PromptInjectionGuard(on_fail="flag"),  # let it through but record the reason
    KeywordBlockGuard(blocked_keywords=["competitor_name"]),
])
result = await chain.check(some_text, context=None)

---

Evaluation

import asyncio
import helix
from helix.eval.suite import EvalSuite
from helix.config import EvalCase

suite = EvalSuite("qa-suite")
suite.add_cases([
    EvalCase(
        name="capital_cities",
        input="What is the capital of France?",
        expected_facts=["Paris"],
        max_cost_usd=0.05,
    ),
    EvalCase(
        name="math",
        input="What is 15% of 240?",
        expected_facts=["36"],
        max_cost_usd=0.05,
    ),
])

async def main():
    agent = helix.Agent(name="Bot", role="Assistant", goal="Answer questions accurately.")
    results = await suite.run(agent, verbose=True)
    print(f"Pass rate:  {results.pass_rate:.0%}")
    print(f"Total cost: ${results.total_cost_usd:.4f}")
    suite.assert_pass_rate(0.90)   # raises AssertionError if below 90%

asyncio.run(main())

The eval suite runs 6 scorers per case: factual accuracy, tool selection, trajectory adherence, cost efficiency, step efficiency, and output quality.

@suite.case decorator:

from helix.eval.suite import EvalSuite
from helix.config import EvalCase

suite = EvalSuite("my-suite")

@suite.case
def capitals():
    return EvalCase(
        input="What is the capital of Germany?",
        expected_facts=["Berlin"],
        max_cost_usd=0.05,
    )

@suite.case
def arithmetic():
    return EvalCase(
        input="What is 25% of 400?",
        expected_facts=["100"],
    )

# suite now has both cases registered; the function name becomes the case name

EvalCase options:

Parameter	Description
input	Task string sent to the agent
expected_facts	Strings that must appear in the output
expected_tools	Tool names the agent is expected to call
expected_trajectory	ExpectedTrajectory for sequence/step constraints
max_steps	Maximum reasoning steps (default 10)
max_cost_usd	Cost cap per case (default 1.00)
pass_threshold	Minimum overall score to pass (default 0.70)
tags	Labels for filtering subsets

---

Framework Adapters

Wrap existing LangChain, CrewAI, or AutoGen code with Helix cost governance:

from langchain_openai import ChatOpenAI
import helix

llm = helix.wrap_llm(ChatOpenAI(model="gpt-4o"), budget_usd=2.00)
# adds budget gate, cost tracking, tracing, and audit log to any LangChain LLM

from langchain.chains import LLMChain
import helix

vchain = helix.from_langchain(LLMChain(...), budget_usd=3.00)
result = await vchain.run(inputs={"input": "Summarise this"})
print(f"Cost: ${vchain.cost_usd:.4f}")

from crewai import Crew
import helix

crew = Crew(agents=[...], tasks=[...])
wrapped = helix.from_crewai(crew, budget_usd=5.00)
result = await wrapped.run(inputs={"topic": "AI trends"})
print(f"Cost: ${wrapped.cost_usd:.4f}")

from autogen import AssistantAgent
import helix

ag_agent = AssistantAgent("assistant", llm_config={...})
wrapped = helix.from_autogen(ag_agent, budget_usd=2.00)
result = await wrapped.run(inputs={"message": "Explain transformers"})

---

CLI

helix doctor                          # check environment and provider keys
helix models                          # list available models with pricing
helix cost --all                      # cost report across all runs
helix trace <run-id>                  # view a run trace
helix trace <run-id> --diff <run-id>  # compare two runs for divergence
helix replay <run-id>                 # interactive failure replay
helix config set KEY value            # set a provider API key

---

Architecture

helix/
├── core/            Agent, ConversableAgent, GroupChat, Task, Pipeline,
│                    Workflow, Team, Session, Tool, StateGraph, AgentPipeline,
│                    Handoff (transfer_to_* tools between agents)
├── presets/         9 ready-made agent factories (web_researcher, coder, writer, …)
├── tools/           13 builtins + MCP client (connect to any MCP server's tools)
├── memory/          Short-term buffer, WAL-backed long-term store, episodic recall,
│                    backends: inmemory (default) | sqlite (persists) | qdrant/pinecone/chroma (not yet implemented)
├── cache/           Semantic cache (tier 1), plan cache (tier 2), prefix cache (tier 3)
├── models/          Router, complexity estimator, 12 provider backends
├── safety/          Cost governor, permission model, guardrails (incl. prompt_injection), HITL, audit log
├── context_engine/  Multi-factor token decay, context compactor, preflight estimator
├── eval/            EvalSuite, 6 scorers, @suite.case decorator, trajectory eval,
│                    regression gate, monitor
├── observability/   Tracer, ghost debug resolver, failure replay
├── adapters/        LangChain, CrewAI, AutoGen + universal LLM wrapper
├── runtime/         Event loop, worker pool, health checks
└── cli/             doctor, models, cost, trace, replay, config, ...

---

Supported Providers

Environment variable	Provider	Notable models	Free tier
GOOGLE_API_KEY	Google Gemini	Gemini 2.5 Flash/Pro, 2.0 Flash	Yes
OPENAI_API_KEY	OpenAI	GPT-4o, GPT-4o-mini, o1, o3, o3-mini	No
ANTHROPIC_API_KEY	Anthropic	Claude Opus 4.6, Sonnet 4.6, Haiku 4.5	No
GROQ_API_KEY	Groq	Llama 3.3-70B, Llama 3.1-8B, Mixtral, Gemma 2	Yes
MISTRAL_API_KEY	Mistral AI	Mistral Large/Small, Codestral, Pixtral	Partial
COHERE_API_KEY	Cohere	Command R+	Partial
TOGETHER_API_KEY	Together AI	200+ open-source models	No
OPENROUTER_API_KEY	OpenRouter	100+ models	Partial
DEEPSEEK_API_KEY	DeepSeek	DeepSeek V3, R1	No
XAI_API_KEY	xAI	Grok	No
PERPLEXITY_API_KEY	Perplexity	Online search models	No
FIREWORKS_API_KEY	Fireworks	Fast open-source inference	No

Set multiple keys and Helix automatically falls back to the next available provider on failure. Ollama (ollama/* or local/*) and any OpenAI-compatible endpoint (Azure, custom base URL) are also supported without environment variable requirements.

---

Contributing

Read CONTRIBUTING.md before opening a PR.

git clone https://github.com/sarcasticdhruv/helix-agent
cd helix-agent
pip install -e ".[dev,gemini]"
pytest tests/

---

Contributors

Name	Role
Dhruv Choudhary	Author and maintainer

---

License

Apache License 2.0. Copyright 2026 Dhruv Choudhary.

See CHANGELOG.md for release history.