以下内容基本上是 AI 生成的,我还没校对,可能质量不高
Promplate¶
Templating framework and agent orchestrator that progressively enhances your prompt engineering workflow with minimal dependencies.
Promplate 是一个专注于开发体验的 Prompting 框架。它既是一个轻量级的模板引擎,也是一个灵活的 LLM 调用 SDK。
核心抽象¶
Promplate 的设计建立在两个极简抽象之上:
抽象一:LLM 是函数¶
# python/promplate/llm/base.py:15-20
class Complete(Protocol):
def __call__(self, prompt, /, **config) -> str: ...
class Generate(Protocol):
def __call__(self, prompt, /, **config) -> Iterable[str]: ... #(1)!
- 流式版本返回可迭代的增量,用
itertools.accumulate累积得到完整输出
任何符合这个签名的函数都可以当作 LLM 使用——OpenAI SDK、Anthropic SDK、甚至一个硬编码的 lambda x: "hello"。
抽象二:Agent 是有限自动机¶
# python/promplate/chain/node.py:107-142
class Interruptible(AbstractNode, Protocol):
def _invoke(self, context: ChainContext, /, complete: Complete | None, callbacks: list[BaseCallback], **config): ...
async def _ainvoke(self, context: ChainContext, /, complete: AsyncComplete | None, callbacks: list[BaseCallback], **config): ...
def _stream(self, context: ChainContext, /, generate: Generate | None, callbacks: list[BaseCallback], **config) -> Iterable: ...
async def _astream(self, context: ChainContext, /, generate: AsyncGenerate | None, callbacks: list[BaseCallback], **config) -> AsyncIterable: ...
Interruptible 定义了状态节点的统一接口。所有节点都遵循相同的生命周期:
# python/promplate/chain/node.py:146-155
def enter(self, context: Context | None, config: Context):
callbacks: list[BaseCallback] = ensure_callbacks(self.callbacks)
for callback in callbacks:
context, config = callback.on_enter(self, context, config)
return context, config, callbacks #(1)!
on_enter/on_leave允许在节点执行前后拦截、修改 context 或 config
在此之上:
Node:单个状态节点(模板 + LLM 调用)Chain:顺序执行的状态序列Loop:带循环的状态链——这就是 AgentJump:控制流转移,允许跳出或跳入任意Interruptible
模板编译器¶
Promplate 的模板引擎会直接生成 Python 代码:
# python/promplate/prompt/template.py:95-117
def compile(self, sync=True, indent_str="\t"):
self._buffer = []
self._ops_stack = []
self._builder = get_base_builder(sync, indent_str) #(1)!
for token in split_template_tokens(self.text):
s_token = token.strip()
if s_token.startswith("{{") and s_token.endswith("}}"):
self._on_eval_token(token)
elif s_token.startswith("{#") and s_token.endswith("#}"):
self._on_exec_token(token)
elif s_token.startswith("{%") and s_token.endswith("%}") and "\n" not in s_token:
self._on_special_token(token, sync)
else:
self._on_literal_token(token)
- 构建器生成
def render(): ...或async def render(): ...函数框架
TemplateCore 的巧妙之处在于错误处理策略:
# python/promplate/prompt/template.py:120-124
error_handling: Literal["linecache", "tempfile", "file"]
if version_info >= (3, 13):
error_handling = "linecache" #(1)!
else:
error_handling = "tempfile" if __debug__ else "file"
- Python 3.13+ 直接注册到
linecache,错误堆栈直接显示模板行号
这段机制确保编译后的模板代码"有文件可循"。出错时,Python 的 traceback 会打印出编译后的源码(而非只显示 <string> 中的错误)。linecache 或临时文件是懒加载的,仅在报错时创建,最小化了运行时开销。
Context 流转¶
ChainContext 继承自 SafeChainMapContext(即 collections.ChainMap 的子类),支持多层上下文叠加:
# python/promplate/chain/node.py:12-53
class ChainContext(SafeChainMapContext):
@overload
def __new__(cls): ...
@overload
def __new__(cls, least: C, *maps: Mapping) -> C: ...
@overload
def __new__(cls, least: MutableMapping | None = None, *maps: Mapping): ...
def __init__(self, least: MutableMapping | None = None, *maps: Mapping):
super().__init__({} if least is None else least, *maps)
@classmethod
def ensure(cls, context):
return context if isinstance(context, cls) else cls(context) #(1)!
- 自动包装普通
dict为ChainContext,确保类型一致性
__result__ 键专门用于存储 LLM 的输出:
# python/promplate/chain/node.py:39-49
@property
def result(self):
return self.__getitem__("__result__")
@result.setter
def result(self, result):
self.__setitem__("__result__", result)
@result.deleter
def result(self):
self.__delitem__("__result__")
这样在 Chain 中后续的节点就能访问前面节点的输出,自然形成数据流。