版本：1.0 日期：2026-04-09 语言规范：AIL v1.0
AIL Python SDK（kzl）是 AIL v1.0 语言规范的 Python 实现。用户直接编写 Python 代码即可使用 AIL 的全部能力，无需学习新语法。
核心设计原则：
- Pythonic 优先——用法尽量符合 Python 惯用法
- Agent 无关——用户传入自己的 agent 实例，SDK 不绑定任何底层模型
- 语义完整——AIL 规范的所有
@操作均有对应实现
pip install ail-sdkimport ail
# 传入用户自己的 agent 实例
my_agent = MyAgent(...) # 任意 agent，SDK 会自动标准化其接口
ai = ail.Runtime(agent=my_agent)SDK 接收任意 agent 对象，在运行时动态检测并标准化其方法。用户的 agent 只需实现 send 方法，其余方法若未实现则由 SDK 提供默认行为。
| 方法 | 签名 | 说明 |
|---|---|---|
send |
(message: str) -> str |
发送消息，返回回复（必须实现） |
reset |
() -> None |
清空全部对话历史 |
remember |
(content: str) -> None |
注入背景信息，不计入对话轮次 |
save |
() -> Any |
保存当前上下文快照，返回快照对象 |
restore |
(snapshot: Any) -> None |
恢复到指定快照状态 |
compact |
() -> None |
压缩上下文：将历史对话摘要化，释放窗口空间，保留语义 |
SDK 在初始化时检查传入 agent 的方法：
- 若 agent 已有同名方法 → 直接使用
- 若 agent 缺少某方法 → SDK 动态附加默认实现（基于
send模拟） send必须存在，否则抛出ail.ConfigError
# 用户 agent 只需实现 send
class MyAgent:
def send(self, message: str) -> str:
# 调用任意底层服务
return my_llm_call(message)
ai = ail.Runtime(agent=MyAgent())
# SDK 自动补全 reset / remember / save / restore / compactresult: str = ai.ask(prompt, **kwargs)prompt：字符串，支持 f-string 或普通字符串**kwargs：可选，用于{var}占位符插值- 返回：
str
result = ai.ask(f"分析任务：{mission}，列出问题")
# 或使用占位符插值
result = ai.ask("分析任务：{mission}，列出问题", mission=mission)result: bool = ai.judge(condition, **kwargs)if ai.judge(f"内容是否已经足够完整：{output}"):
breakresult = ai.pick(instruction, options: list, **kwargs)返回值类型与 options 元素类型一致。
action = ai.pick("根据用户意图选择下一步", options=["搜索", "直接回答", "追问用户"])steps: list[str] = ai.plan(goal, **kwargs)steps = ai.plan(f"完成目标：{goal}，可用工具：{tools}")
result = ""
for step in steps:
result = ai.ask(f"执行「{step}」，已有结论：{result}")result = ai.extract(text, type, hint="", **kwargs)type：目标类型，支持 Python 类型注解、自定义类、dictschema、ail.sorted(T)hint：可选的提取指引
# 提取为基本类型
keywords = ai.extract(text, type=list[str], hint="提取关键词列表")
score = ai.extract(text, type=float, hint="提取评分")
# 提取为自定义类（需用 @ai.type 注册，见第六节）
info = ai.extract(analysis, type=QueryInfo)
# 提取为匿名 schema
info = ai.extract(analysis, type={"intent": str, "keywords": list[str]})
# 语义排序
ranked = ai.extract(docs, type=ail.sorted(Document), hint=f"按与问题「{query}」的相关性排序")
top5 = ranked[:5]result = ai.eval(content, criterion="", type=None, **kwargs)# 单维度，返回 float
score = ai.eval(answer, "回答与问题的相关性")
# 多维度，返回对象
quality = ai.eval(answer, type={"relevance": float, "completeness": float})
if quality.relevance < 0.7:
answer = ai.ask(f"改进回答，当前相关性：{quality.relevance}")ai.validate(content, condition, **kwargs)
# 不满足时抛出 ail.ValidationError(reason)ai.validate(result, "内容必须包含具体数字和来源")配合 ai.retry 使用时，ValidationError 触发整块重试（见 4.10）。
result: str = ai.act(instruction, max_steps=1, **kwargs)max_steps：允许 AI 连续调用工具的最大轮数，默认1- 工具集从已注册的
@ai.tool函数自动生成描述
result = ai.act(f"根据用户问题选择合适的工具：{query}")
# 允许多轮工具调用
result = ai.act(f"完成复杂任务：{task}", max_steps=5)with ai.context(system="", model="") as ctx:
...块内所有 ai.* 操作共享同一段对话历史。支持嵌套。
with ai.context(system="你是一个任务规划专家") as ctx:
step1 = ai.ask(f"分析任务：{mission}")
step2 = ai.ask("针对上述问题给出方案") # AI 能看到 step1
ctx.remember("用户偏好简洁回答") # 注入背景
ctx.reset() # 清空历史
snapshot = ctx.save() # 保存快照
ctx.restore(snapshot) # 恢复快照
ctx.compact() # 压缩历史离开 with 块后，上下文自动销毁，后续调用重新独立。
with ai.retry(max=3):
result = ai.ask(...)
ai.validate(result, "条件")
# ValidationError 触发重试，达到上限后向上抛出with ai.timeout("30s"): # 支持 s / m / h
result = ai.ask(...)
# 超时抛出 ail.TimeoutError直接使用 Python 原生语法：
try:
result = ai.ask(deep_prompt)
except ail.AIError:
result = ai.ask(basic_prompt)ail.AIError 是所有 SDK 异常的基类，包括 ValidationError、TimeoutError、CancelError。
内部使用 ThreadPoolExecutor，兼容任意同步 agent。
with ai.parallel() as p:
r1 = p.ask("从技术角度分析：{content}", content=content)
r2 = p.ask("从商业角度分析：{content}", content=content)
r3 = p.ask("从用户角度分析：{content}", content=content)
# with 块结束时并行执行，阻塞直到全部完成
summary = ai.ask(f"综合三个角度给出结论：{r1} {r2} {r3}")parallel 块内可以调用任意 ai.* 操作，但不能修改块外的变量（与 AIL 规范一致）。
extra = ai.ask_user("请补充更多背景信息") # 阻塞，返回 str
ai.confirm(f"即将向 {recipient} 发送报告，确认？") # 阻塞，拒绝时抛出 ail.CancelError
ai.show(f"已找到 {len(docs)} 篇文档，正在生成...") # 非阻塞，无返回memory 的存储与检索由传入的 agent 自身管理，SDK 提供统一调用接口。
ai.memory.save(content, key="", tags=[]) # 存储，key 可选
ai.memory.get(key) # 精确读取 → str
ai.memory.search(query, top_k=5) # 语义搜索 → list[str]ai.memory.save("用户偏好简洁回答", key="user_pref", tags=["preference"])
pref = ai.memory.get("user_pref")
related = ai.memory.search("RAG 相关历史", top_k=3)@ai.tool
def vector_search(query: str, top_k: int) -> list:
"""Search vector database for semantically similar documents."""
...
@ai.tool
def send_email(to: str, subject: str, body: str) -> bool:
"""Send an email to the specified recipient."""
...注册后可直接调用，也可被 ai.act 自动使用。
@ai.skill
def summarizer(text: str) -> str:
return ai.ask(f"将以下内容总结为要点：{text}")
@ai.skill
def rag_agent(query: str) -> tuple[str, list[str]]:
...
return answer, citationsskill 内部可以包含完整的 ai.* 调用，对外表现为普通函数。
@ai.plugin
class Calendar:
def get_events(self, date: str) -> list:
...
def create_event(self, title: str, time: str) -> bool:
...
# 使用
events = Calendar.get_events(date="2026-04-10")
Calendar.create_event(title="周会", time="14:00")用于 ai.extract 的 type= 参数：
@ai.type
class Document:
id: str
title: str
content: str
score: float = 0.0
@ai.type
class QueryInfo:
intent: str
keywords: list[str]
multi_step: bool标准 Python dataclass 风格，SDK 在运行时自动序列化为 JSON schema 注入 prompt。
ail.AIError # 所有异常的基类
├── ail.ValidationError # @validate 不满足条件
├── ail.TimeoutError # timeout 超时
└── ail.CancelError # @confirm 被用户拒绝
每个 ai.* 操作在调用 agent 时使用固定的 prompt 框架包装用户输入，约束 AI 的输出格式。
system: You are a helpful assistant executing a task as instructed.
user: {user_prompt}
system: You are a precise evaluator. You must respond with a single boolean value.
user:
[Condition]
{user_condition}
[Output Format]
Reply with only: true or false
No explanation. No punctuation. Lowercase only.
system: You are a decision-making assistant. Select exactly one option.
user:
[Instruction]
{user_instruction}
[Options]
1. option_a
2. option_b
...
[Output Format]
Reply with only the number of your chosen option.
system: You are a planning assistant. Break down goals into clear, executable steps.
user:
[Goal]
{user_goal}
[Output Format]
Return a numbered list of steps, one per line. No explanation.
system: You are a precise data extractor. Return valid JSON only.
user:
[Source Text]
{source_text}
[Extraction Task]
{hint}
[Target Schema]
{auto_generated_schema}
[Output Format]
Return only a valid JSON object matching the schema. Raw JSON, no markdown.
system: You are an objective evaluator. Score from 0.0 to 1.0.
user:
[Content]
{content}
[Dimensions]
- relevance: ...
- completeness: ...
[Output Format]
Return only JSON: {"relevance": 0.85, "completeness": 0.60}
system: You are a strict content validator.
user:
[Content]
{content}
[Condition]
{user_condition}
[Output Format]
Line 1: pass or fail
Line 2: one-sentence reason (required on fail)
system: You are an autonomous agent. Select the most appropriate tool and call it.
user:
[Task]
{user_instruction}
[Available Tools]
1. vector_search(query: str, top_k: int) -> list
{docstring}
2. send_email(to: str, subject: str, body: str) -> bool
{docstring}
[Output Format]
{"tool": "<name>", "args": {"param": value}}
Raw JSON only.
import ail
# 初始化
ai = ail.Runtime(agent=MyAgent())
# 注册工具
@ai.tool
def vector_search(query: str, top_k: int) -> list:
"""Search vector database."""
...
@ai.tool
def keyword_search(query: str) -> list:
"""Keyword-based document search."""
...
@ai.tool
def rerank(query: str, docs: list) -> list:
"""Rerank documents by relevance."""
...
# 注册数据类型
@ai.type
class Document:
id: str
title: str
content: str
score: float = 0.0
@ai.type
class QueryInfo:
intent: str
keywords: list[str]
multi_step: bool
# RAG Agent
def rag_answer(user_query: str, max_rounds: int = 3) -> tuple[str, list[str]]:
with ai.context() as ctx:
# 1. 理解问题
analysis = ai.ask(f"分析用户问题：{user_query}，输出核心意图、关键词列表、是否需要多步推理")
info = ai.extract(analysis, type=QueryInfo)
# 2. 并行检索
vec_query = ai.ask(f"将意图「{info.intent}」改写为适合向量检索的语句")
kw_query = ai.ask(f"从关键词「{info.keywords}」中提取最重要的 3 个")
with ai.parallel() as p:
vec_docs = p.task(vector_search, vec_query, top_k=10)
kw_docs = p.task(keyword_search, kw_query)
top_docs = rerank(user_query, vec_docs + kw_docs)[:5]
# 3. 补充检索
for _ in range(max_rounds):
if ai.judge(f"以下文档能否充分回答「{user_query}」：{top_docs}"):
break
gap = ai.ask(f"当前文档还缺少哪些信息才能回答「{user_query}」")
new_docs = vector_search(ai.ask(f"根据缺口生成补充检索语句：{gap}"), top_k=5)
top_docs = rerank(user_query, top_docs + new_docs)[:5]
# 4. 推理生成
if info.multi_step:
steps = ai.plan(f"针对「{user_query}」制定推理步骤，参考：{top_docs}")
result = ""
for step in steps:
result = ai.ask(f"执行「{step}」，已有结论：{result}，参考：{top_docs}")
final_context = result
else:
final_context = ai.ask(f"整理文档中与「{user_query}」相关的核心内容：{top_docs}")
# 5. 生成答案，校验质量
generate_prompt = f"""
基于以下内容回答用户问题。
问题：{user_query}
参考内容：{final_context}
要求：有理有据，引用来源，不要编造
"""
with ai.retry(max=3):
answer = ai.ask(generate_prompt)
ai.validate(answer, "回答必须基于参考内容，不得出现参考内容中没有的事实")
quality = ai.eval(answer, type={"relevance": float, "completeness": float})
if quality.relevance < 0.7 or quality.completeness < 0.7:
answer = ai.ask(f"改进以上回答，当前评分：{quality}")
# 6. 提取引用，保存记忆
citations = ai.extract(answer, type=list[str], hint="提取引用来源")
ai.memory.save(answer, key=f"rag:{user_query[:20]}", tags=["history"])
return answer, citations| 问题 | 决策 |
|---|---|
| 实现形式 | Python SDK，Pythonic 风格 |
| Agent 绑定 | 运行时传入，接口动态标准化 |
| 模型调用 | 通过 agent.send() 抽象，不绑定任何底层 LLM |
| 变量插值 | 支持 f-string 和 {key} + kwargs 两种方式 |
loop |
用户自己写 for/while，SDK 提供 ai.judge |
fallback |
Python 原生 try/except ail.AIError |
parallel |
ThreadPoolExecutor，兼容任意同步 agent |
| 工具注册 | @ai.tool / @ai.skill / @ai.plugin 装饰器 |
| 类型注册 | @ai.type 装饰器，运行时序列化为 JSON schema |
| memory 后端 | 由 agent 自身管理，SDK 提供统一调用接口 |
act 多轮 |
max_steps 参数控制，默认 1 |
| 异常体系 | ail.AIError 为基类，三个子类 |