Describe your business logic in natural language. Let AI execute it.
Version: v1.0
Traditional code only handles deterministic logic: if score > 5 — the outcome is predictable. But in AI workflows, many decisions are inherently semantic: "Is this answer good enough?" "Is the task complete?" — these judgments need AI to make, not a program.
AIL is designed for exactly this. It has only one rule:
@ → AI executes (semantic, non-deterministic)
→ code runs (deterministic, identical to Python)
This example has AI iteratively refine a vague task description until there are no more issues:
def upgrade_mission(mission: str) -> str:
with context(system="you are a task planning expert") as ctx:
prompt analyze = """
analyze the task "{mission}", list what is unclear or needs improvement
"""
prompt solve = """
for the issues above, provide answers and improvement suggestions
"""
output = @ask(analyze) # AI executes, returns str
loop max=10 until @judge("the following content has no unresolved issues: {output}"):
output = @ask(solve) # AI judges whether to continue
return @ask("summarize the discussion above, output the final task prompt")
One read and you understand it:
- Inside a conversation (
with context), have AI analyze the task and find issues - Repeatedly have AI resolve issues until AI judges there are none (
loop until @judge) - Summarize and output
All AI operations begin with @. The most common:
| Operation | Purpose | Returns |
|---|---|---|
@ask(prompt) |
have AI execute a task | str |
@judge("condition") |
have AI make a yes/no judgment | bool |
@pick("instruction", options=[...]) |
have AI choose from options | option type |
@plan("goal") |
have AI decompose a goal into steps | list[str] |
@extract(text, type=T) |
extract structured data from text | T |
@eval(content, "criterion") |
have AI score content | float (0–1) |
@validate(content, "condition") |
assert — raises on failure | — / exception |
@act("instruction") |
AI autonomously selects and calls a tool | str |
@ask_user("prompt") |
ask the user, wait for input | str |
@confirm("description") |
request user confirmation, stop on reject | — / exception |
@show("content") |
display content to user (non-blocking) | — |
Use prompt to define prompts for AI. {variable} is automatically resolved from the current scope at call time:
# global definition, reused across functions
prompt common_summary = """
summarize the following into a bullet list: {content}
"""
# local definition, used nearby (recommended)
def process(query: str) -> str:
prompt do_analyze = """
deeply analyze the following question and give a structured answer.
question: {query}
"""
return @ask(do_analyze) # query is automatically taken from scope
Short prompts can be inlined directly without defining a prompt:
result = @ask("summarize in one sentence: {text}")
Variable interpolation rules:
| Variable type | Rendered as |
|---|---|
str / int / float / bool |
converted to string directly |
list[str] |
numbered list: 1. a\n2. b |
list[type] |
numbered list with fields expanded per item |
type object |
fields expanded line by line |
dict |
key: value line by line |
with context() as ctx: creates a conversation block. All @ operations inside share the same conversation history — AI can see every prior step:
with context() as ctx:
step1 = @ask(prompt_a)
step2 = @ask(prompt_b) # AI can see step1
step3 = @ask(prompt_c) # AI can see both previous steps
Specify model and role at creation:
with context(model="claude-opus") as ctx: # specify model
with context(system="you are a medical advisor") as ctx: # set role
with context(model="claude-opus", system="...") as ctx: # both
Context object operations:
ctx.remember("user prefers concise answers") # inject background (not counted in turns)
ctx.reset() # clear conversation history
snapshot = ctx.save() # save current state
ctx.restore(snapshot) # restore to a state
When you don't need multi-turn conversation, use @ask directly — each call is independent:
summary = @ask("summarize in one sentence: {text}") # no context, independent call
The most basic AI operation:
result = @ask(my_prompt) # reference a prompt template (recommended)
result = @ask("summarize in one sentence: {text}") # short inline prompt
result = @ask(my_prompt, model="claude-opus") # temporarily switch model
Used directly in conditions and loops:
if @judge("are there still unresolved issues in output"):
output = @ask(solve_issues)
ready = @judge("are the docs sufficient to answer '{query}': {docs}")
Have AI break a large goal into a step list, then execute step by step:
steps = @plan("complete goal: {goal}, available tools: {tools}")
# pass intermediate results between steps (common pattern)
result = ""
for step in steps:
result = @ask("execute '{step}', prior conclusions: {result}, reference: {docs}")
Extract the structure you need from AI's text output:
# extract to custom type
info = @extract(analysis, type=QueryInfo)
# extract to primitive types
keywords = @extract(text, "extract keyword list", type=list[str])
score = @extract(text, "extract score", type=float)
flag = @extract(text, "requires multi-step reasoning", type=bool)
# semantic sort (returns list re-ranked by relevance)
ranked = @extract(docs, "sort by relevance to '{query}'", type=sorted[Document])
top5 = ranked[:5]
Have AI score content quality, returns a float 0–1:
# single dimension
score = @eval(answer, "relevance of answer to question")
# multi-dimension (use type= to specify fields, returns object)
quality = @eval(answer, type={"relevance": float, "completeness": float})
if quality.relevance < 0.7:
answer = @ask("improve the answer, current relevance score: {quality.relevance}")
Assert content meets a condition. Raises on failure (combine with retry for automatic retries):
# standalone: stop if not met
@validate(result, "content must include specific numbers and sources")
# with retry: retry the whole block if not met
retry max=3:
result = @ask(generate)
@validate(result, "content must be over 200 words")
@judge used directly in if, can be mixed with regular Python conditions:
if @judge("are there still unresolved issues in output"):
output = @ask(solve_issues)
elif len(results) == 0: # regular condition, same as Python
output = @ask(try_another_way)
else:
pass
AI re-judges whether to exit after each iteration. max= is a safety cap — always set it:
loop max=10 until @judge("is the task complete: {output}"):
output = @ask(continue_task)
for step in steps:
result = @ask("execute step: {step}")
while not ready:
output = @ask(continue_task)
ready = @judge("is it ready: {output}")
Several independent tasks run simultaneously, saving time:
# parallel tools
vec_docs, kw_docs = parallel:
vec_docs = vector_search(vec_query, top_k=10)
kw_docs = keyword_search(kw_query)
# parallel AI tasks
r1, r2, r3 = parallel:
r1 = @ask("analyze technically: {content}")
r2 = @ask("analyze commercially: {content}")
r3 = @ask("analyze from user perspective: {content}")
summary = @ask("synthesize three perspectives for a final conclusion: {r1} {r2} {r3}")
Note:
parallelblocks can read outer variables, but cannot modify them.
AI calls can fail, or return results that don't meet requirements. AIL provides three mechanisms:
Automatically retries the whole block when @validate fails:
retry max=3:
report = @ask(generate_report)
@validate(report, "report must include conclusion, data, and sources")
timeout 30s:
result = @ask(complex_analysis)
# raises TimeoutError after 30 seconds; supports s / m / h
Any exception (including timeout, ValidationError) triggers fallback:
try:
result = @ask(deep_analysis, model="claude-opus")
fallback:
result = @ask(basic_analysis)
try:
retry max=3:
timeout 20s:
result = @ask(generate_report)
@validate(result, "report must be complete")
fallback:
result = @ask(basic_report)
AIL supports three kinds of external capabilities, each with distinct semantics:
| Type | Description | Use case |
|---|---|---|
use tool |
deterministic function, no AI, no state | search, compute, file I/O |
use skill |
sub-agent, can contain a full AI workflow inside | complex subtasks, reusable agents |
use plugin |
external service with persistent state | databases, calendars, email services |
use tool vector_search(query: str, top_k: int) -> list[Document]
use tool send_email(to: str, subject: str, body: str) -> bool
# call like a regular function after declaring
docs = vector_search("deep learning", top_k=10)
ok = send_email("user@example.com", "Report", report)
Can contain a full AI workflow inside, called like a regular function from outside:
use skill summarizer(text: str) -> str
use skill rag_agent(query: str) -> (str, list[str])
summary = summarizer(text=article)
answer, citations = rag_agent(query=user_query)
Has persistent state, accessed via .method():
use plugin calendar
use plugin database as db
events = calendar.get_events(date="2026-04-10")
calendar.create_event(title="Weekly sync", time="14:00")
users = db.query("SELECT * FROM users WHERE active = 1")
db.save(new_record)
Register a toolset and let AI decide which one to call:
use tools [search, calculator, read_file]
result = @act("choose the right tool and execute based on user question: {query}")
memory is a built-in object that lets agents remember information across sessions. Three operations:
# store (key optional; required for exact retrieval; tags for categorization)
memory.save("user prefers concise bullet answers", key="user_preference", tags=["preference"])
memory.save(summary, key="last_summary", tags=["history"])
memory.save(user_profile, tags=["user"]) # no key, can only be found via search
# exact retrieval (by key)
preference = memory.get("user_preference")
# semantic search (AI-driven, returns top N most relevant)
related = memory.search("prior discussions about RAG", top_k=5)
Using memory in prompts:
preference = memory.get("user_preference")
related = memory.search("history related to current topic", top_k=3)
prompt personalized_answer = """
user preference: {preference}
relevant history: {related}
answer the question based on the following: {query}
"""
result = @ask(personalized_answer)
When an agent needs to pause and wait for the user:
# ask the user (blocking, wait for input)
extra = @ask_user("please provide more background information")
# request user confirmation (blocking, stops execution on reject)
@confirm("about to send report to {recipient}, proceed?")
# show intermediate result to user (non-blocking, no wait)
@show("found {len(docs)} documents, generating answer...")
Use type to define custom structures. Fields support default values and inline comments:
type Document:
id: str # unique document identifier
title: str # document title
content: str # body text
score: float = 0.0 # relevance score, 0–1
type Point:
x: float
y: float
Access is identical to Python:
doc = docs[0]
title = doc.title
top5 = docs[:5]
Identical to Python syntax. Supports type annotations, default parameters, multiple return values:
# single return value
def summarize(text: str, max_words: int = 200) -> str:
prompt do_summarize = """
summarize the following text to no more than {max_words} words: {text}
"""
return @ask(do_summarize)
# multiple return values
def analyze(query: str) -> (str, list[str]):
with context() as ctx:
answer = @ask("answer the question: {query}")
keywords = @extract(answer, "extract keywords", type=list[str])
return answer, keywords
# calling
summary = summarize("a very long piece of text...", max_words=100)
answer, keywords = analyze("what is a vector database?")
A complete RAG (Retrieval-Augmented Generation) agent demonstrating most AIL features.
use tool vector_search(query: str, top_k: int) -> list[Document]
use tool keyword_search(query: str) -> list[Document]
use tool rerank(query: str, docs: list[Document]) -> list[Document]
type Document:
id: str # unique document identifier
title: str # document title
content: str # body text
score: float = 0.0 # relevance score, 0–1
type QueryInfo:
intent: str # user's core intent
keywords: list[str] # keyword list
multi_step: bool # requires multi-step reasoning
def rag_answer(user_query: str, max_rounds=3) -> (str, list[str]):
with context() as ctx:
# step 1: understand the user's question
prompt analyze_query = """
analyze user question: {user_query}
output: 1. core intent 2. keyword list 3. whether multi-step reasoning is needed
"""
analysis = @ask(analyze_query)
info = @extract(analysis, type=QueryInfo)
# step 2: parallel retrieval (vector + keyword simultaneously)
intent = info.intent
keywords = info.keywords
vec_query = @ask("rewrite intent '{intent}' as a vector search query")
kw_query = @ask("extract the 3 most important terms from keywords '{keywords}'")
vec_docs, kw_docs = parallel:
vec_docs = vector_search(vec_query, top_k=10)
kw_docs = keyword_search(kw_query)
top_docs = rerank(user_query, vec_docs + kw_docs)[:5]
# step 3: supplement retrieval when docs are insufficient
prompt find_gap = """
what information is still missing from the current docs to answer '{user_query}'
"""
loop max=max_rounds until @judge("can these docs sufficiently answer '{user_query}': {top_docs}"):
gap = @ask(find_gap)
new_docs = vector_search(@ask("generate a supplemental search query for gap: {gap}"), top_k=5)
top_docs = rerank(user_query, top_docs + new_docs)[:5]
# step 4: generate answer (single or multi-step reasoning)
if info.multi_step:
steps = @plan("devise reasoning steps for '{user_query}', reference: {top_docs}")
result = ""
for step in steps:
result = @ask("execute '{step}', prior conclusions: {result}, reference: {top_docs}")
final_context = result
else:
final_context = @ask("organize core content from docs relevant to '{user_query}': {top_docs}")
# step 5: generate final answer, validate quality
prompt generate_answer = """
answer the user question based on the following.
question: {user_query}
reference: {final_context}
requirements: well-reasoned, cite sources, no fabrication
"""
retry max=3:
answer = @ask(generate_answer)
@validate(answer, "answer must be grounded in reference content, no facts absent from it")
quality = @eval(answer, type={"relevance": float, "completeness": float})
if quality.relevance < 0.7 or quality.completeness < 0.7:
answer = @ask("improve the answer above, current scores: {quality}")
# step 6: extract citations, save to memory
citations = @extract(answer, "extract cited sources, cross-reference with docs", type=list[str], context=top_docs)
memory.save(answer, key=f"rag:{user_query[:20]}", tags=["history"])
return answer, citations
AI operations
| Operation | Returns | Description |
|---|---|---|
@ask(prompt) |
str |
have AI execute a task |
@judge("condition") |
bool |
yes/no judgment |
@pick("instruction", options=[]) |
option type | choose from options |
@plan("goal") |
list[str] |
decompose goal into steps |
@extract(x, ..., type=T) |
T |
extract structured data; sorted[T] for semantic sort |
@eval(x, "criterion") |
float |
score 0–1; type={} for multi-dimension |
@validate(x, "condition") |
— / exception | assert; stop or retry on failure |
@act("instruction") |
str |
AI autonomously selects and calls a tool |
@ask_user("prompt") |
str |
ask the user (blocking) |
@confirm("description") |
— / exception | request user confirmation (blocking) |
@show("content") |
— | display to user (non-blocking) |
Control flow
| Statement | Description |
|---|---|
if @judge(...): |
AI conditional branch, can mix with regular conditions |
loop max=N until @judge(...): |
AI semantic loop, re-judges each iteration |
while / for |
regular loops, same as Python |
parallel: |
run independent tasks concurrently |
with context(...) as ctx: |
create shared conversation context |
Reliability
| Statement | Description |
|---|---|
retry max=N: |
retry whole block until @validate passes |
timeout Ns: |
timeout (s/m/h) |
try: / fallback: |
fallback on error; any exception triggers fallback |
Declarations and extensions
| Statement | Description |
|---|---|
prompt name = """...""" |
define a prompt template (global or local) |
type Name: |
define a data structure |
use tool func(...) -> T |
declare a deterministic tool function |
use skill func(...) -> T |
declare a sub-agent skill |
use plugin name |
declare a stateful external plugin |
use tools [...] |
register a toolset for @act |
def func(...) -> T: |
define a function |
Memory system
| Operation | Description |
|---|---|
memory.save(content, key="", tags=[]) |
store memory; key optional, required for exact retrieval |
memory.get("key") |
exact retrieval by key |
memory.search("description", top_k=N) |
semantic search |