taolib.symphony.config

目录

taolib.symphony.config#

Symphony 配置子包。

提供工作流加载、配置模型、TOML 解析、配置合并与文件监视功能。

使用方式：

from taolib.symphony.config import (

WorkflowDefinition, load_workflow, SymphonyConfig, resolve_config, WorkflowWatcher,

)

Submodules#

Classes#

WorkflowDefinition

解析后的工作流定义。

AgentConfig

Agent 并发与重试配置。

CodexConfig

Codex app-server 配置。

HooksConfig

工作区钩子配置。

PollingConfig

轮询配置。

ServerConfig

HTTP Server 扩展配置。

SymphonyConfig

Symphony 编排服务完整配置。

TrackerConfig

问题跟踪器配置。

WorkerConfig

SSH Worker 扩展配置。

WorkspaceConfig

工作区配置。

WorkflowWatcher

WORKFLOW.md 文件监视器。

Functions#

load_workflow(→ WorkflowDefinition)

从文件加载 WORKFLOW.md 定义。

deep_merge(→ dict)

深度合并两个字典。

resolve_config(...)

组装最终配置。

resolve_env_vars(→ dict)

替换配置中的 $VAR_NAME 环境变量引用。

resolve_paths(→ dict)

解析配置中的路径值。

load_toml(→ dict)

从 symphony.toml 文件加载 [defaults] 段。

Package Contents#

class taolib.symphony.config.WorkflowDefinition#

解析后的工作流定义。

config#

YAML 前置数据根对象（映射）。

prompt_template#

前置数据之后的 Markdown 正文（已去除首尾空白）。

config: dict#
prompt_template: str#
taolib.symphony.config.load_workflow(path: pathlib.Path) → WorkflowDefinition#

从文件加载 WORKFLOW.md 定义。

解析规则： - 如果文件以 --- 开头，解析至下一个 --- 为 YAML 前置数据。 - 剩余内容为 prompt_template。 - 如果没有前置数据，整个文件作为 prompt_template，config 为空映射。 - YAML 前置数据必须解码为映射，否则抛出 WorkflowLoadError。

参数:

path -- WORKFLOW.md 文件路径。

返回:

解析后的 WorkflowDefinition。

抛出:

WorkflowLoadError -- 文件无法读取或前置数据不是映射。

taolib.symphony.config.deep_merge(base: dict, override: dict) → dict#

深度合并两个字典。

override 中的值覆盖 base 中的同名键。如果两边的值都是字典， 则递归合并。base 和 override 均不被修改。

参数:

  • base -- 基础配置字典。

  • override -- 覆盖配置字典。

返回:

合并后的新字典。

taolib.symphony.config.resolve_config(cli_args: dict, toml_path: pathlib.Path | None, workflow_path: pathlib.Path) → taolib.symphony.config.schema.SymphonyConfig#

组装最终配置。

按优先级从低到高合并配置源： 1. Pydantic 默认值 2. symphony.toml [defaults] 段 3. WORKFLOW.md YAML 前置数据 4. 环境 $VAR 替换 5. 路径解析 6. CLI 参数覆盖

参数:

  • cli_args -- CLI 参数字典，例如 {"port": 8080, "logs_root": Path("./log")}。

  • toml_path -- symphony.toml 文件路径，为 None 则跳过。

  • workflow_path -- WORKFLOW.md 文件路径。

返回:

校验后的 SymphonyConfig 实例。

抛出:
taolib.symphony.config.resolve_env_vars(config: dict) → dict#

替换配置中的 $VAR_NAME 环境变量引用。

仅对值为 $VAR_NAME 形式的字符串进行替换。 如果环境变量不存在或为空字符串，则保留原始 $VAR_NAME 引用。

参数:

config -- 原始配置字典。

返回:

替换环境变量后的配置字典。

taolib.symphony.config.resolve_paths(config: dict, base_dir: pathlib.Path) → dict#

解析配置中的路径值。

对已知的路径字段进行 ~ 展开和相对路径解析： - ~ 展开为用户主目录。 - 相对路径基于 base_dir 解析为绝对路径。 - 仅处理已知的路径字段，不重写 URI 或任意命令字符串。

参数:

  • config -- 配置字典。

  • base_dir -- 相对路径的基准目录（通常为 WORKFLOW.md 所在目录）。

返回:

路径解析后的配置字典。

class taolib.symphony.config.AgentConfig#

Bases: pydantic.BaseModel

Agent 并发与重试配置。

max_concurrent_agents: int#
max_turns: int#
max_retry_backoff_ms: int#
max_concurrent_agents_by_state: dict[str, int]#
classmethod _normalize_state_keys(v: dict[str, int]) → dict[str, int]#

将状态键归一化为小写，并过滤无效条目。

class taolib.symphony.config.CodexConfig#

Bases: pydantic.BaseModel

Codex app-server 配置。

command: str = 'codex app-server'#
approval_policy: str | None = None#
thread_sandbox: str | None = None#
turn_sandbox_policy: str | None = None#
turn_timeout_ms: int#
read_timeout_ms: int#
stall_timeout_ms: int#
class taolib.symphony.config.HooksConfig#

Bases: pydantic.BaseModel

工作区钩子配置。

after_create: str | None = None#
before_run: str | None = None#
after_run: str | None = None#
before_remove: str | None = None#
timeout_ms: int#
class taolib.symphony.config.PollingConfig#

Bases: pydantic.BaseModel

轮询配置。

interval_ms: int#
class taolib.symphony.config.ServerConfig#

Bases: pydantic.BaseModel

HTTP Server 扩展配置。

port: int | None = None#
bind: str = '127.0.0.1'#
class taolib.symphony.config.SymphonyConfig#

Bases: pydantic.BaseModel

Symphony 编排服务完整配置。

tracker: TrackerConfig#
polling: PollingConfig#
workspace: WorkspaceConfig#
hooks: HooksConfig#
agent: AgentConfig#
codex: CodexConfig#
worker: WorkerConfig#
server: ServerConfig#
class taolib.symphony.config.TrackerConfig#

Bases: pydantic.BaseModel

问题跟踪器配置。

kind: Literal['linear'] = 'linear'#
endpoint: str = 'https://api.linear.app/graphql'#
api_key: str = ''#
project_slug: str = ''#
active_states: list[str]#
terminal_states: list[str]#
class taolib.symphony.config.WorkerConfig#

Bases: pydantic.BaseModel

SSH Worker 扩展配置。

ssh_hosts: list[str]#
max_concurrent_agents_per_host: int | None = None#
class taolib.symphony.config.WorkspaceConfig#

Bases: pydantic.BaseModel

工作区配置。

root: pathlib.Path#
taolib.symphony.config.load_toml(path: pathlib.Path) → dict#

从 symphony.toml 文件加载 [defaults] 段。

参数:

path -- symphony.toml 文件路径。

返回:

[defaults] 段的字典，如果文件中没有 [defaults] 段则返回空字典。

抛出:

ConfigError -- 文件无法读取或 TOML 解析失败。

class taolib.symphony.config.WorkflowWatcher(workflow_path: pathlib.Path, on_reload_success: OnReloadSuccess | None = None, on_reload_error: OnReloadError | None = None)#

WORKFLOW.md 文件监视器。

基于 watchdog 监视指定的工作流文件，变更时重新加载并验证配置。 验证失败则保持最后有效配置，不会使服务崩溃。

使用方式：

watcher = WorkflowWatcher(

workflow_path=Path("WORKFLOW.md"), on_reload_success=my_success_handler, on_reload_error=my_error_handler,

) watcher.start() # ... 服务运行 ... watcher.stop()

_workflow_path#
_handler#
_observer: watchdog.observers.Observer | None = None#
start() → None#

启动文件监视。

stop() → None#

停止文件监视。

property last_valid_config: taolib.symphony.config.schema.SymphonyConfig | None#

返回最后有效的配置。

目录