Runtime#

Source: openprogram/agentic_programming/runtime.py

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

Class: 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 自动算出。

在 @agentic_function 内部调用时:

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

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

一个 @agentic_function 可以多次调用 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) -> Any

exec() 的异步版本,具有相同的 response_format=None 文本返回以及结构化 Python JSON 返回/异常合同。内部调用 _async_call();call= 函数可同步或异步, 默认 provider 路径使用同一 AgentSession structured lifecycle。timeout_s / on_retry 语义与 exec() 相同;重试用 asyncio.sleep 休眠,外部取消能生效。 它没有公开的工具循环参数。

_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。

使用方式#

方式一:传入 call 函数#

from openprogram import agentic_function
from openprogram.agentic_programming.runtime import Runtime

def my_llm(content, model="sonnet", response_format=None):
    # 把 content 转成你的 provider 格式,发请求
    texts = [b["text"] for b in content if b["type"] == "text"]
    return call_my_api("\n".join(texts), model=model)

runtime = Runtime(call=my_llm, model="sonnet")

@agentic_function
def observe(task):
    """Look at the screen."""
    return runtime.exec(content=[
        {"type": "text", "text": f"Find: {task}"},
        {"type": "image", "path": "screenshot.png"},
    ])

方式二:子类化#

class MyAnthropicRuntime(Runtime):
    def __init__(self, api_key, model="sonnet"):
        super().__init__(model=model)
        self.client = anthropic.Anthropic(api_key=api_key)

    def _call(self, content, model="sonnet", response_format=None):
        messages_content = []
        for block in content:
            if block["type"] == "text":
                messages_content.append({"type": "text", "text": block["text"]})
        response = self.client.messages.create(
            model=model, max_tokens=1024,
            messages=[{"role": "user", "content": messages_content}],
        )
        return response.content[0].text

runtime = MyAnthropicRuntime(api_key="sk-...", model="claude-sonnet-4-6")

多个 Runtime 共存#

fast = Runtime(call=gemini_call, model="gemini-2.5-flash")
strong = Runtime(call=claude_call, model="sonnet")

@agentic_function
def observe(task):
    """Quick observation with cheap model."""
    return fast.exec(content=[...])

@agentic_function
def plan(goal):
    """Complex planning with strong model."""
    return strong.exec(content=[...])

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

重试的边界#

max_retries 只处理 API 层面的瞬态故障(网络超时、速率限制等)。如果是函数本身的逻辑或输出格式有问题,重试解决不了——直接修改函数代码,参见 skills/agentic-programming/SKILL.md。