aacf 0.8.0

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.

---

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(who="Translator", what="Translate Chinese to English")
def translate(text: str):
    pass

print(translate(text="Hello World"))
# -> 你好世界

---

Core Ideas

Five-tuple DSL. Reduce prompts to who / where / what / why / how. Each AI node is an atomic function with a clear role.

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 -e .

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(who="Title Writer", what="Generate 3 article titles for a topic", stream=True)
def title_generator(topic: str):
    pass

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

@app.node(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(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(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(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(who="Analyzer", what="Analyze text", cache_enabled=True, cache_ttl=300)
def analyzer(text: str):
    pass

---

CLI

aacf init my_project        # Initialize project
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

---

API Reference

@app.node() Parameters

Parameter	Type	Required	Description
who	str	Yes	Agent role
what	str	Yes	Core task
where	str		Business context
why	str		Execution intent
how	str | list		Steps or constraints
module	list[Callable]		Sub-nodes for smart routing
out	str		Output format requirements
stream	bool		True returns Generator
format	str		"json" enables JSON mode
branches	dict[str, Callable]		Conditional branch targets
cache_enabled	bool		Enable result caching (default False)
cache_ttl	int		Cache TTL in seconds (default 0)
max_retries	int		Max retry attempts (default 3)
retry_delay	float		Delay between retries in seconds (default 1.0)
timeout	int		Execution timeout in seconds (default 0, no timeout)

LLMConfig

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

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

---

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

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

---

Installation

git clone https://github.com/yourusername/aacf.git
cd aacf
pip install -e .

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

---

Documentation

For detailed documentation, see Wiki.md.

---

License

GPL-3.0. See LICENSE.