用自然语言描述业务逻辑，让 AI 帮你执行。
文档版本：v1.0
传统代码只能处理确定性逻辑：if score > 5，结果是可预期的。但在 AI 工作流里，很多判断本质上是语义性的——"这个回答够好了吗？""任务完成了吗？"——这些判断需要 AI 来做，不是程序。
AIL 就是为此设计的。它的规则只有一条：
有 @ → 交给 AI 执行（语义判断，结果由 AI 决定）
没有 @ → 程序直接执行（确定性逻辑，和 Python 完全一致）
下面这个例子让 AI 把一个模糊的任务描述反复打磨，直到没有问题为止：
def upgrade_mission(mission: str) -> str:
with context(system="你是一个任务规划专家") as ctx:
prompt analyze = """
分析任务「{mission}」，列出你不清楚或需要优化的问题
"""
prompt solve = """
针对上述问题，给出解答和优化方案
"""
output = @ask(analyze) # AI 执行任务，返回 str
loop max=10 until @judge("以下内容不再有待解决的问题：{output}"):
output = @ask(solve) # AI 判断是否继续
return @ask("总结以上讨论，输出最终的任务提示词")
读一遍就能理解：
- 在一段对话里（
with context），让 AI 分析任务、找问题 - 反复让 AI 解决问题，直到 AI 判断没问题了（
loop until @judge） - 总结输出
所有 AI 操作都以 @ 开头。常用的：
| 操作 | 用途 | 返回值 |
|---|---|---|
@ask(prompt) |
让 AI 执行任务 | str |
@judge("条件") |
让 AI 做是/否判断 | bool |
@pick("说明", options=[...]) |
让 AI 从选项中选一个 | 选项类型 |
@plan("目标") |
让 AI 把目标拆成步骤 | list[str] |
@extract(text, type=T) |
从文本里提取结构化数据 | T |
@eval(content, "标准") |
让 AI 打分 | float (0~1) |
@validate(content, "条件") |
断言，不满足就报错 | 无/异常 |
@act("说明") |
AI 自主选择并调用工具 | str |
@ask_user("提示") |
向用户提问，等待输入 | str |
@confirm("描述") |
请求用户确认，拒绝时终止 | 无/异常 |
@show("内容") |
展示内容给用户（不阻塞） | 无 |
用 prompt 定义给 AI 用的提示词，{变量} 在调用时自动从当前作用域取值：
# 全局定义，跨函数复用
prompt common_summary = """
请将以下内容总结为要点列表：{content}
"""
# 局部定义，就近使用（推荐）
def process(query: str) -> str:
prompt do_analyze = """
深入分析以下问题，给出结构化答案。
问题：{query}
"""
return @ask(do_analyze) # query 自动从作用域取值
简短的提示词可以直接内联，不必定义 prompt：
result = @ask("一句话总结：{text}")
变量插值规则：
| 变量类型 | 渲染结果 |
|---|---|
str / int / float / bool |
直接转字符串 |
list[str] |
编号列表：1. a\n2. b |
list[type] |
每项展开字段的编号列表 |
type 对象 |
字段逐行展开 |
dict |
key: value 逐行展开 |
with context() as ctx: 创建一个对话区块，块内所有 @ 操作共享同一段对话历史——AI 能看到前面每一步的内容：
with context() as ctx:
step1 = @ask(prompt_a)
step2 = @ask(prompt_b) # AI 能看到 step1 的内容
step3 = @ask(prompt_c) # AI 能看到前两步
创建时可以指定模型和角色：
with context(model="claude-opus") as ctx: # 指定模型
with context(system="你是一名医疗顾问") as ctx: # 设置角色
with context(model="claude-opus", system="...") as ctx: # 同时设置
上下文对象的操作：
ctx.remember("用户偏好简洁回答") # 注入背景信息（不计入对话轮次）
ctx.reset() # 清空对话历史
snapshot = ctx.save() # 保存当前状态
ctx.restore(snapshot) # 恢复到某个状态
不需要多轮对话时，直接用 @ask 即可，每次独立调用：
summary = @ask("一句话总结：{text}") # 无上下文，独立调用
最基础的 AI 操作：
result = @ask(my_prompt) # 引用 prompt 模板（推荐）
result = @ask("一句话总结：{text}") # 短提示词内联
result = @ask(my_prompt, model="claude-opus") # 临时切换模型
直接用在条件和循环里：
if @judge("output 中还有未解决的问题"):
output = @ask(solve_issues)
ready = @judge("文档是否足够回答「{query}」：{docs}")
让 AI 把一个大目标分解成步骤列表，再逐步执行：
steps = @plan("完成目标：{goal}，可用工具：{tools}")
# 步骤间传递中间结果（常用模式）
result = ""
for step in steps:
result = @ask("执行「{step}」，已有结论：{result}，参考：{docs}")
从 AI 的文本输出里提取你需要的结构：
# 提取为自定义类型
info = @extract(analysis, type=QueryInfo)
# 提取为基本类型
keywords = @extract(text, "提取关键词列表", type=list[str])
score = @extract(text, "提取评分", type=float)
flag = @extract(text, "是否需要多步推理", type=bool)
# 语义排序（返回按相关性重排后的列表）
ranked = @extract(docs, "按与问题「{query}」的相关性排序", type=sorted[Document])
top5 = ranked[:5]
让 AI 对内容质量评分，返回 0~1 的浮点数：
# 单维度
score = @eval(answer, "回答与问题的相关性")
# 多维度（用 type= 指定字段，返回对象）
quality = @eval(answer, type={"relevance": float, "completeness": float})
if quality.relevance < 0.7:
answer = @ask("改进回答，当前相关性评分：{quality.relevance}")
断言内容满足条件，不满足时抛出错误（配合 retry 使用可以自动重试）：
# 单独使用：不满足就终止
@validate(result, "内容必须包含具体数字和来源")
# 配合 retry：不满足就重试整个块
retry max=3:
result = @ask(generate)
@validate(result, "内容必须超过 200 字")
@judge 直接用在 if 里，可与普通 Python 条件混用：
if @judge("output 中还有未解决的问题"):
output = @ask(solve_issues)
elif len(results) == 0: # 普通条件，同 Python
output = @ask(try_another_way)
else:
pass
每轮结束后 AI 重新判断是否退出。max= 是安全上限，建议始终设置：
loop max=10 until @judge("任务是否已完成：{output}"):
output = @ask(continue_task)
for step in steps:
result = @ask("执行步骤：{step}")
while not ready:
output = @ask(continue_task)
ready = @judge("是否已准备好：{output}")
几个互不依赖的任务同时跑，节省时间：
# 工具并行
vec_docs, kw_docs = parallel:
vec_docs = vector_search(vec_query, top_k=10)
kw_docs = keyword_search(kw_query)
# AI 任务并行
r1, r2, r3 = parallel:
r1 = @ask("从技术角度分析：{content}")
r2 = @ask("从商业角度分析：{content}")
r3 = @ask("从用户角度分析：{content}")
summary = @ask("综合三个角度，给出最终结论：{r1} {r2} {r3}")
注意：
parallel块内可以读取外部变量，但不能修改外部变量。
AI 调用可能失败，或返回不满足要求的结果。AIL 提供三种应对机制：
@validate 不通过时，自动重试整个块：
retry max=3:
report = @ask(generate_report)
@validate(report, "报告必须包含结论、数据和来源")
timeout 30s:
result = @ask(complex_analysis)
# 超过 30 秒抛出 TimeoutError，支持 s / m / h
任何异常（包括超时、ValidationError）都触发 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, "报告必须完整")
fallback:
result = @ask(basic_report)
AIL 支持三种外部能力，语义不同，各司其职：
| 类型 | 说明 | 适用场景 |
|---|---|---|
use tool |
确定性函数，无 AI，无状态 | 搜索、计算、文件读写 |
use skill |
子 Agent，内部可有完整 AI 工作流 | 复杂子任务、可复用的 Agent |
use plugin |
带持久状态的外部服务 | 数据库、日历、邮件服务 |
use tool vector_search(query: str, top_k: int) -> list[Document]
use tool send_email(to: str, subject: str, body: str) -> bool
# 声明后像普通函数一样调用
docs = vector_search("深度学习", top_k=10)
ok = send_email("user@example.com", "报告", report)
内部可以有完整的 AI 工作流，对外像普通函数一样调用：
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)
有持久状态，通过 .方法名() 调用：
use plugin calendar
use plugin database as db
events = calendar.get_events(date="2026-04-10")
calendar.create_event(title="周会", time="14:00")
users = db.query("SELECT * FROM users WHERE active = 1")
db.save(new_record)
注册工具集后，由 AI 决定调用哪个：
use tools [search, calculator, read_file]
result = @act("根据用户问题选择合适的工具并执行：{query}")
memory 是内置对象，让 Agent 能跨会话记住信息。三个操作：
# 存储（key 可选，有 key 才能精确取回；tags 用于分类）
memory.save("用户喜欢简洁的列表式回答", key="user_preference", tags=["preference"])
memory.save(summary, key="last_summary", tags=["history"])
memory.save(user_profile, tags=["user"]) # 没有 key，只能用 search 找到
# 精确取回（按 key）
preference = memory.get("user_preference")
# 语义搜索（AI 驱动，返回最相关的 N 条）
related = memory.search("和 RAG 相关的历史讨论", top_k=5)
在提示词里使用记忆：
preference = memory.get("user_preference")
related = memory.search("与当前话题相关的历史", top_k=3)
prompt personalized_answer = """
用户偏好：{preference}
相关历史：{related}
基于以下内容回答问题：{query}
"""
result = @ask(personalized_answer)
Agent 需要暂停等待用户时：
# 向用户提问（阻塞，等待输入）
extra = @ask_user("请补充更多背景信息")
# 请求用户确认（阻塞，拒绝时终止执行）
@confirm("即将向 {recipient} 发送报告，是否继续？")
# 展示中间结果给用户（非阻塞，不等待）
@show("已找到 {len(docs)} 篇文档，正在生成回答...")
用 type 定义自定义结构，字段支持默认值和行内注释：
type Document:
id: str # 文档唯一标识
title: str # 文档标题
content: str # 正文内容
score: float = 0.0 # 相关性评分，0~1
type Point:
x: float
y: float
访问方式与 Python 完全一致：
doc = docs[0]
title = doc.title
top5 = docs[:5]
与 Python 语法完全一致，支持类型注解、默认参数、多返回值：
# 单返回值
def summarize(text: str, max_words: int = 200) -> str:
prompt do_summarize = """
将以下文本总结为不超过 {max_words} 字：{text}
"""
return @ask(do_summarize)
# 多返回值
def analyze(query: str) -> (str, list[str]):
with context() as ctx:
answer = @ask("回答问题：{query}")
keywords = @extract(answer, "提取关键词", type=list[str])
return answer, keywords
# 调用
summary = summarize("很长的一段文本...", max_words=100)
answer, keywords = analyze("什么是向量数据库？")
下面是一个完整的 RAG（检索增强生成）Agent，展示 AIL 大多数特性的综合运用。
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 # 文档唯一标识
title: str # 文档标题
content: str # 正文内容
score: float = 0.0 # 相关性评分，0~1
type QueryInfo:
intent: str # 用户的核心意图
keywords: list[str] # 关键词列表
multi_step: bool # 是否需要多步推理
def rag_answer(user_query: str, max_rounds=3) -> (str, list[str]):
with context() as ctx:
# 第一步：理解用户的问题
prompt analyze_query = """
分析用户问题：{user_query}
请输出：1.核心意图 2.关键词列表 3.是否需要多步推理
"""
analysis = @ask(analyze_query)
info = @extract(analysis, type=QueryInfo)
# 第二步：并行检索（向量 + 关键词同时进行）
intent = info.intent
keywords = info.keywords
vec_query = @ask("将意图「{intent}」改写为适合向量检索的语句")
kw_query = @ask("从关键词「{keywords}」中提取最重要的 3 个")
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]
# 第三步：文档不够时，继续补充检索
prompt find_gap = """
当前文档还缺少哪些信息才能回答「{user_query}」
"""
loop max=max_rounds until @judge("以下文档能否充分回答「{user_query}」：{top_docs}"):
gap = @ask(find_gap)
new_docs = vector_search(@ask("根据缺口生成补充检索语句：{gap}"), top_k=5)
top_docs = rerank(user_query, top_docs + new_docs)[:5]
# 第四步：生成答案（支持单步和多步推理）
if info.multi_step:
steps = @plan("针对「{user_query}」制定推理步骤，参考：{top_docs}")
result = ""
for step in steps:
result = @ask("执行「{step}」，已有结论：{result}，参考：{top_docs}")
final_context = result
else:
final_context = @ask("整理文档中与「{user_query}」相关的核心内容：{top_docs}")
# 第五步：生成最终答案，校验质量
prompt generate_answer = """
基于以下内容回答用户问题。
问题：{user_query}
参考内容：{final_context}
要求：有理有据，引用来源，不要编造
"""
retry max=3:
answer = @ask(generate_answer)
@validate(answer, "回答必须基于参考内容，不得出现参考内容中没有的事实")
quality = @eval(answer, type={"relevance": float, "completeness": float})
if quality.relevance < 0.7 or quality.completeness < 0.7:
answer = @ask("改进以上回答，使其更相关、更完整，当前评分：{quality}")
# 第六步：提取引用来源，保存到记忆
citations = @extract(answer, "提取引用来源，与以下文档对照", type=list[str], context=top_docs)
memory.save(answer, key=f"rag:{user_query[:20]}", tags=["history"])
return answer, citations
AI 操作
| 操作 | 返回 | 说明 |
|---|---|---|
@ask(prompt) |
str |
让 AI 执行任务 |
@judge("条件") |
bool |
让 AI 做是/否判断 |
@pick("说明", options=[]) |
选项类型 | 让 AI 从选项中选择 |
@plan("目标") |
list[str] |
让 AI 拆解目标为步骤 |
@extract(x, ..., type=T) |
T |
提取结构化数据；sorted[T] 语义排序 |
@eval(x, "标准") |
float |
让 AI 打分（0~1）；type={} 多维度 |
@validate(x, "条件") |
无/异常 | 断言，不满足则终止或重试 |
@act("说明") |
str |
AI 自主选择工具执行 |
@ask_user("提示") |
str |
向用户提问（阻塞） |
@confirm("描述") |
无/异常 | 请求用户确认（阻塞） |
@show("内容") |
无 | 展示内容给用户（非阻塞） |
控制流
| 语句 | 说明 |
|---|---|
if @judge(...): |
AI 条件分支，可与普通条件混用 |
loop max=N until @judge(...): |
AI 语义循环，每轮重新判断 |
while / for |
普通循环，同 Python |
parallel: |
并行执行独立任务 |
with context(...) as ctx: |
创建共享对话上下文 |
可靠性
| 语句 | 说明 |
|---|---|
retry max=N: |
整块重试，直到 @validate 通过 |
timeout Ns: |
超时（s/m/h） |
try: / fallback: |
出错降级，任何异常触发 fallback |
声明与扩展
| 语句 | 说明 |
|---|---|
prompt name = """...""" |
定义提示词模板（全局或局部） |
type Name: |
定义数据结构 |
use tool func(...) -> T |
声明确定性工具函数 |
use skill func(...) -> T |
声明子 Agent skill |
use plugin name |
声明带状态的外部插件 |
use tools [...] |
注册工具集供 @act 使用 |
def func(...) -> T: |
定义函数 |
记忆系统
| 操作 | 说明 |
|---|---|
memory.save(内容, key="", tags=[]) |
存储记忆，key 可选，有 key 才能精确取回 |
memory.get("key") |
精确读取记忆 |
memory.search("描述", top_k=N) |
语义搜索记忆 |