运行时 API#

Source: openprogram/agentic_programming/runtime.py

LLM 运行时。封装 LLM provider,自动从 session DAG 算上下文、调用 LLM、把回复写回 DAG。

Runtime 类#

class Runtime(call=None, model="default", max_retries=None, api_key=None, skills=None)

构造参数#

属性#

方法#

exec()#

Runtime.exec(content, context=None, response_format=None, model=None,
             tools=None, toolset=None, tools_source=None, tools_allow=None,
             tools_deny=None, tool_choice="auto", parallel_tool_calls=True,
             max_iterations=20, choices=None, timeout_s=None, on_retry=None,
             web_search=False, stream_fn=None) -> Any

调用 LLM,上下文从 session DAG 自动算出。

在 Agent 方法或受管函数作用域内调用时：

1. 从当前函数的 DAG 节点出发,render_context 按 expose / render_range 算出本次要读哪些历史节点
2. render_dag_messages 把这些节点渲染成 messages
3. 调用 _call() 发送请求
4. 回复写进 exec() 开始时打开的那个 llm 节点

没安装 DAG store 时(standalone 脚本、无 dispatcher):content 包成单条 user message 作单轮调用发送,不做任何记录。

Agent 方法或函数作用域可以多次调用 exec(),每次都是 DAG 上的一个新 llm 节点。

参数#

Content block 格式#

{"type": "text",  "text": "Find the login button."}
{"type": "image", "path": "screenshot.png"}
{"type": "image", "data": "<base64>", "mime_type": "image/png"}
{"type": "video", "path": "clip.mp4"}
{"type": "audio", "path": "recording.wav"}

媒体块给 path(自动读取并 base64 编码,mime 类型按扩展名猜)或内联 data + mime_type 都行。text 块可带 "role": "system" 汇入 system prompt;text/image 块接受 cache_control 做 provider prompt caching。未知块类型静默跳过。

返回值#

response_format=None 时返回 LLM 回复文本 str。传结构化 response_format 时返回经过本地校验的 Python JSON 值(dict、list、标量或 None)。带 choices 时返回解析后的决策结果(选中函数的返回值,或选中值本身)。

异常#

• RuntimeError — runtime 已关闭(调用过 close())
• TypeError / NotImplementedError — 立即抛出,从不重试(编程错误:调用签名不对、没配置 provider)
• LLMError — 重试耗尽或遇到不可重试错误时抛出,结构化字段含 reason / retryable / http_status / retry_after_s / attempts / elapsed_s / provider / model 等
• StructuredOutputSchemaError — 公共 schema 或包络非法;稳定 code="invalid_schema"
• StructuredOutputUnsupportedError — 没有已验证且无损的 provider/fallback 合同;稳定 code="unsupported"
• StructuredOutputValidationError — JSON 非法、schema 不匹配或隐藏提交合同失败;稳定 code 为 invalid_json、validation_failed、missing_submission、mixed_submission
• StructuredOutputGenerationError — provider 拒绝或未完整生成;稳定 code 为 refusal、incomplete

全部 structured-output 异常都提供稳定 .code 和有界 .issues 字段。 provider 候选正文不属于公共诊断合同。

async_exec()#

await Runtime.async_exec(content, context=None, response_format=None, model=None,
                         timeout_s=None, on_retry=None, tools=None, toolset=None,
                         max_iterations=None) -> Any

exec() 的异步版本,具有相同的 response_format=None 文本返回以及结构化 Python JSON 返回/异常合同。内部调用 _async_call();call= 函数可同步或异步, 默认 provider 路径使用同一 AgentSession structured lifecycle。timeout_s / on_retry 语义与 exec() 相同;重试用 asyncio.sleep 休眠,外部取消能生效。 它接受同步工具循环选项，包括 tools、toolset、max_iterations 和工具授权过滤。

_call()#

Runtime._call(content, model="default", response_format=None) -> Any

实际调用一次 LLM 的方法(不含重试——重试循环在 exec() 里包着)。默认实现在配置了 provider 模型或 call= 函数时走 provider 层(AgentSession),否则抛 NotImplementedError。子类化时重写此方法。

参数#

返回值#

普通调用返回原始文本;结构化格式激活时返回已校验的 Python JSON 值。

_async_call()#

await Runtime._async_call(content, model="default", response_format=None) -> Any

_call() 的异步版本。子类化时重写此方法以支持异步 provider。

close()#

释放资源、结束会话;close() 之后 exec() 抛 RuntimeError。Runtime 也是上下文管理器(with Runtime(...) as rt: 退出时自动关闭)。子类重写它清理 provider 专属资源。

向用户提问#

有前端会话连着时,runtime 可以在函数中途阻塞等用户输入:runtime.ask(prompt, options=..., multi=..., questions=[...], timeout=300.0, default=None)(一题或多题一屏),runtime.confirm(prompt, default=False)(是/否),runtime.form(prompt, fields)(多字段表单)。runtime.can_ask() 报告当前有没有人能回答(headless 跑时为 False)。用户拒绝抛 UserDeclined;超时给了 default 就返回它,否则抛 AskTimeout。

使用方式#

宿主提供 Runtime，Agent 自动继承它。普通方法不需要装饰器或手动 Context 管理。

from openprogram import Agent, Runtime


def my_llm(content, model="host-model", response_format=None):
    texts = [block["text"] for block in content if block["type"] == "text"]
    return call_my_api("\n".join(texts), model=model)

runtime = Runtime(call=my_llm, model="host-model")

class Observer(Agent):
    tools = []

    def observe(self, task: str):
        """Identify the requested screen element."""
        return self([
            {"type": "text", "text": f"Find: {task}"},
            {"type": "image", "path": "screenshot.png"},
        ])

observer = Observer(runtime=runtime)
try:
    answer = observer.observe("login button")
finally:
    runtime.close()

Agent 与 Context API 说明配置、显式会话与作用域。方法配置继续兼容。

Retry 机制#

exec() 和 async_exec() 内置自动重试,用于处理 LLM API 的临时性错误(网络超时、速率限制、服务器错误等)。

配置#

# 默认:max_retries=None → 读环境变量 OPENPROGRAM_MAX_RETRIES,没设则为 6
rt = Runtime(call=my_llm)

# 不重试(失败即抛异常)
rt = Runtime(call=my_llm, max_retries=1)

# 多次重试(适用于不稳定的 API)
rt = Runtime(call=my_llm, max_retries=5)

行为规则#

错误报告格式#

当所有重试耗尽时,抛出的 LLMError 包含每次尝试的错误信息,结构化字段(reason / retryable / http_status / attempts / elapsed_s 等)可直接读取:

LLMError: exec() failed after 3 attempt(s):
Attempt 1: ConnectionError: timeout
Attempt 2: RateLimitError: 429 Too Many Requests
Attempt 3: ConnectionError: timeout

重试的边界#

瞬态传输失败、结构化输出修正，以及启用工具时的异常重复输出，共用本次调用的失败恢复额度。一次模型请求在传输失败后重试成功，会归还所用额度，因此传输恢复按单次请求计限，而不是按整个 Agent 循环计限。重复响应在同一 Agent 上下文内重新生成，保留已完成的工具结果及 Runtime 回执，不设置整体 Token 或执行时长上限。显式调用限制、取消、权限失败和恢复额度耗尽仍然有效。重新生成响应不代表可以重复外部写入；写入结果不确定时必须先检查实际记录。函数自身的逻辑错误仍需修改代码。