aacf 0.11.1

Agentic AI Compiler Framework — AI workflow pre-compilation and highly controllable scheduling execution engine

AACF - Agentic AI Compiler Framework

English | 中文

A Python framework for building LLM-driven agent pipelines through decorators, dependency analysis, and DAG-based scheduling.

---

Design Philosophy

You think you're thinking freely. Actually you're thinking freely within the correct range I designed. This is not restriction. This is the most elegant liberation.

Weak models do powerful things through clear rules. Strong models do rigorous things through constrained freedom.

Most frameworks try to make AI more autonomous. AACF goes the opposite direction: it makes AI thinking explicitly rule-bound — not through temperature knobs, but through structured declarations the model internalizes as its own reasoning.

Level 1 — Chaos:          Model does anything. Unpredictable. Uncontrollable.
Level 2 — Rigid:          Temperature/top_p suppression. Stable but unexplained.
Level 3 — Rule-internalized (AACF):  Model "feels" free, but every choice
                          is within your designed boundaries. Predictable.
                          Explainable. Controllable.

This is the "open design" principle: the model believes it is making intelligent decisions, while in reality all decisions unfold within the space you explicitly defined through who / what / where / why / how. Not a cage — a chessboard. The rules don't limit the game; they make the game possible.

---

What It Does

You declare AI nodes with a decorator. AACF handles prompt construction, LLM calls, dependency analysis, and execution scheduling.

from aacf import AACF, LLMConfig

app = AACF(__name__, config=LLMConfig(
    model="qwen2.5-7b-instruct",
    url="http://127.0.0.1:8080/v1/chat/completions",
))

@app.node("translate").who("Translator").what("Translate Chinese to English")
def translate(text: str):
    pass

print(translate(text="Hello World"))
# -> Hello World

---

Core Ideas

Rule-internalized freedom. The five-tuple DSL (who / where / what / why / how) constrains model thinking within designed boundaries — not through temperature, but through explicit rules the model internalizes as its own reasoning.

Human-controlled flow. LLMs act as classifiers within nodes, not as controllers. Developers use native Python (if/elif/for) to direct data flow.

Precompilation. Before execution, AACF analyzes parameter names, infers dependencies, builds a DAG, and generates a topological execution plan.

Atomic execution nodes. Each node is independently schedulable, retryable, and cacheable. Failed nodes retry with configurable backoff.

Rust-style errors. ExecutionResult makes error handling explicit and mandatory. No silent failures.

OpenAI-compatible. Switch between cloud APIs and local models by changing a URL. No code changes.

Explicit code override. Function body is pass -> framework calls LLM. Function body has code -> your code runs. Switch back to pass anytime.

---

Quick Start

pip install aacf

agents.py -- Define nodes:

from aacf import AACF, LLMConfig

app = AACF(__name__, config=LLMConfig(
    model="qwen2.5-7b-instruct",
    url="http://127.0.0.1:8080/v1/chat/completions",
    language="en",  # "zh" or "en"
))

@app.node("title_generator").who("Title Writer").what("Generate 3 article titles for a topic").stream(True)
def title_generator(topic: str):
    pass

@app.node("article_writer").who("Article Writer").what("Write a 200-word article from a title")
def article_writer(title: str):
    pass

@app.node("content_router").who("Content Director").what("Route requests to the right node").module([title_generator, article_writer])
def content_router(user_req: str):
    pass

main.py -- Call them:

from agents import title_generator, article_writer, content_router

# Streaming
for chunk in title_generator(topic="AI in daily life"):
    print(chunk, end="", flush=True)

# Regular call
print(article_writer(title="When AI learned to cook"))

# Smart routing -- auto-dispatches to the best node
print(content_router(user_req="Write me an article about quantum computing"))

---

Precompilation

AACF analyzes node dependencies before execution:

app.compile()                    # Build DAG and execution plan
app.get_execution_order()        # -> ["title_generator", "article_writer", ...]
app.get_parallel_groups()        # -> [["title_generator"], ["article_writer"], ...]
app.get_dependency_graph()       # -> {"article_writer": {"title_generator"}, ...}

Dependency inference works by matching parameter names to node names. If article_writer(title) has a parameter title and there is a node called title_generator, the dependency is inferred when names align.

---

Features

Streaming Output

@app.node("writer").who("Writer").what("Write a short story").stream(True)
def writer(topic: str):
    pass

for chunk in writer(topic="Cyberpunk city"):
    print(chunk, end="", flush=True)

Structured JSON

@app.node("extractor").who("Data Extractor").what("Extract person info").format("json")
def extractor(text: str):
    pass

import json
data = json.loads(extractor(text="Li Lei, 28, engineer"))

Explicit Code Override

@app.node("calculator").who("Calculator").what("Calculate result")
def calculator(expression: str):
    # Your code runs instead of the default LLM call
    return str(eval(expression))

Error Handling

from aacf import PipelineError

try:
    results = app.run_pipeline(inputs={...})
except PipelineError as e:
    print(f"Pipeline failed: {e}")

DAG Visualization

from aacf import DAGVisualizer

visualizer = DAGVisualizer(app)
visualizer.generate_html("dag.html")  # Interactive HTML

Caching

@app.node("analyzer").who("Analyzer").what("Analyze text").cache(ttl=300)
def analyzer(text: str):
    pass

---

CLI

aacf init my_project            # Initialize project (creates venv + installs aacf)
aacf init my_project --no-venv  # Initialize project (skip venv, instant)
aacf run main.py                # Run script
aacf sync .                     # Inject docstrings into source
aacf watch .                    # Watch and auto-inject
aacf doc aacf --port 8080       # API doc server

---

MCP Server

AACF provides an MCP (Model Context Protocol) server for AI-assisted development. AI clients like Claude Desktop can use AACF tools to help you build and manage projects.

# Install with MCP support
pip install aacf[mcp]

# Start MCP server (stdio mode)
aacf-mcp

Client Configuration

Qoder (.qoder/mcp.json):

{
  "mcpServers": {
    "aacf": {
      "command": "python",
      "args": ["-m", "aacf_mcp"]
    }
  }
}

Claude Desktop (claude_desktop_config.json):

{
  "mcpServers": {
    "aacf": {
      "command": "python",
      "args": ["-m", "aacf_mcp"]
    }
  }
}

Use python -m aacf_mcp instead of aacf-mcp for better compatibility across environments.

**Available MCP Tools:**

| Category | Tools |
|----------|-------|
| Project | `init_project`, `read_project`, `validate_project` |
| Nodes | `create_node`, `list_nodes`, `get_node_info`, `configure_node` |
| Pipeline | `compile_pipeline`, `get_dependency_graph`, `get_execution_order`, `get_parallel_groups`, `run_pipeline` |

---

## API Reference

### `@app.node()` Chainable API

```python
# Basic usage
@app.node("name").who("Role").what("Task")
def my_node(param: str):
    pass

# Full chainable configuration
@app.node("name") \
    .who("Role") \
    .where("Context") \
    .what("Task") \
    .why("Intent") \
    .how("Steps") \
    .stream(True) \
    .format("json") \
    .cache(ttl=300) \
    .retry(max_attempts=3, delay=1.0) \
    .timeout(30)
def my_node(param: str):
    pass

Chainable Methods

Method	Description
.who(role)	Set agent role
.where(context)	Set business context
.what(task)	Set core task
.why(intent)	Set execution intent
.how(steps)	Set steps or constraints
.module([nodes])	Set sub-nodes for smart routing
.out(format)	Set output format requirements
.stream(True)	Enable streaming output
.format("json")	Enable JSON mode
.cache(ttl=300)	Enable caching with TTL
.retry(max_attempts=3, delay=1.0)	Configure retry behavior
.timeout(30)	Set execution timeout

LLMConfig

config = LLMConfig(
    model="qwen2.5-7b-instruct",
    url="http://localhost:8080/v1/chat/completions",
    api_key="",  # optional, omit for local models
    temperature=0.7,
    max_tokens=1024,
    language="en",
)

# Derive new config (original unchanged)
hot_config = config(temperature=1.2)

Uses OpenAI-compatible Chat Completions API (POST /v1/chat/completions). Works with OpenAI, DeepSeek, Azure OpenAI, vLLM, Ollama, LM Studio, LocalAI, and more. See Wiki.md for full configuration guide.

---

Project Structure

aacf/
  __init__.py        # Exports: AACF, LLMConfig, ExecutionResult, ...
  core.py            # Engine: config, HTTP client, decorator
  compiler.py        # Dependency analysis, DAG, atomic scheduler, error handling
  visualize.py       # Interactive HTML DAG visualization (pyvis)
  cli.py             # CLI commands
  _messages.py       # Bilingual prompt templates

aacf_mcp/            # MCP Server (optional)
  __init__.py        # Exports: create_server
  server.py          # FastMCP server with stdio transport
  tools/
    nodes.py         # Node management tools
    pipeline.py      # Pipeline analysis tools
    project.py       # Project management tools

examples/
  agents.py          # Demo: content creation assistant
  main.py            # Demo: invocation entry point

---

Installation

# From PyPI (recommended)
pip install aacf

# With MCP server support
pip install aacf[mcp]

# From source
git clone https://github.com/Roxy-DD/aacf-py.git
cd aacf-py
pip install -e .

Python >= 3.10. Core dependencies: typer, rich. Optional: pyvis (visualization), mcp (MCP server).

---

Documentation

For detailed documentation, see Wiki.md.

---

License

GPL-3.0. See LICENSE.