SDK 指南 · 安装、配置与最佳实践

安装

SDK 需要 Python 3.10 或更高版本。通过 pip 安装最新版本。

pip install ieda-agent-sdk

# Verify installation
python -c "import ieda; print(ieda.__version__)"
# Output: 0.2.0

# Install with optional dependencies
pip install ieda-agent-sdk[dev]      # Development tools: pytest, mypy, ruff
pip install ieda-agent-sdk[tracing]  # Performance tracing support
pip install ieda-agent-sdk[all]      # Everything

配置

SDK 通过环境变量、配置文件或代码初始化进行配置。优先级：代码参数 > 环境变量 > 配置文件。

环境变量

# Required
export IEDA_HOST=localhost          # iEDA Platform 地址
export IEDA_PORT=9090              # Interface gateway 端口

# Optional
export IEDA_PROTOCOL=python        # 默认协议: python | mcp | tcl
export IEDA_TIMEOUT_MS=30000       # 默认超时（毫秒）
export IEDA_TRACING=false          # 是否启用性能追踪
export IEDA_MAX_RETRIES=3          # 默认重试次数
export IEDA_LOG_LEVEL=INFO         # 日志级别: DEBUG | INFO | WARN | ERROR

配置文件

# ~/.ieda/config.yaml
host: localhost
port: 9090
protocol: python
timeout_ms: 30000
tracing: false
max_retries: 3
log_level: INFO
connection:
  pool_size: 4
  keepalive_sec: 60
  reconnect_delay_ms: 1000

连接设置

from ieda import AgentClient

# Minimal configuration
client = AgentClient()
client.connect()

# Full configuration
client = AgentClient(
    host="ieda-platform.local",
    port=9090,
    protocol="python",
    timeout_ms=60000,
    tracing=True,
    log_level="DEBUG",
    connection_pool_size=8,
    keepalive_sec=30,
)
client.connect()
assert client.is_connected()

错误处理

SDK 定义了分层级的异常类型——从网络层到工具层，每种异常都有明确的含义和推荐的恢复策略。

异常层级

IEDAError                    # 基础异常
├── ConnectionError          # 网络层：无法连接 / 连接断开
│   ├── ConnectionRefused
│   └── ConnectionTimeout
├── ValidationError           # 参数层：schema 校验失败
│   ├── MissingField
│   ├── InvalidType
│   └── OutOfRange
├── ToolError                 # 工具层：工具执行失败
│   ├── ToolNotFound
│   └── ToolExecutionError
├── TimeoutError              # 超时
├── GateFailure               # 门禁层：evaluation gate 未通过
├── CASConflict               # 并发层：CAS commit 冲突
└── RateLimitError            # 限流层：调用频率超限

重试策略

from ieda import retry, RetryConfig

# Built-in retry with exponential backoff
result = retry(
    lambda: client.call("iSTA.analyze", snap=snap.id),
    config=RetryConfig(
        max_attempts=3,
        backoff_ms=1000,
        backoff_multiplier=2.0,
        max_backoff_ms=30000,
        retry_on={TimeoutError, ConnectionError},
    ),
)

# Manual retry with CAS conflict handling
for attempt in range(3):
    try:
        result = client.call("iDB.commit", expected_head=head, proposal=prop)
        break
    except CASConflict as e:
        print(f"Rebasing to {e.conflict_with} (attempt {attempt+1})")
        prop = rebase_proposal(prop, e.conflict_with)

降级模式

# Graceful degradation when a tool is unavailable
try:
    timing = client.call("iSTA.analyze", snap=snap.id)
except ToolNotFound:
    logging.warning("iSTA not available — falling back to basic timing")
    timing = basic_timing_estimate(design)
except RateLimitError:
    logging.warning("Rate limited — using cached result")
    timing = cache.get("timing", snap.id)

最佳实践

批量操作

# Avoid: N individual calls
for cell in critical_cells:
    result = client.call("iDB.inspect", cell_id=cell.id)  # Latency adds up

# Prefer: single batch call
results = client.call("iDB.batch_inspect", cell_ids=[c.id for c in critical_cells])

结果缓存

from ieda import cached_call

# Cache snapshot-based results — same snap = same result
timing = cached_call(
    client, "iSTA.analyze",
    snap=snap.id,
    ttl_sec=300,  # Cache for 5 minutes
)

连接池复用

# Production setup — reuse connection pool
client = AgentClient(connection_pool_size=8)
client.connect()

# Use context manager for automatic cleanup
with AgentClient(connection_pool_size=4) as client:
    design = client.load_design("aes_core", tech="sky130")
    result = client.call("iPL.place", design=design)
# Connection pool automatically closed

优雅关闭

import signal

client = AgentClient()
client.connect()

def shutdown(signum, frame):
    print("Shutting down...")
    client.disconnect()
    print("Disconnected.")

signal.signal(signal.SIGINT, shutdown)
signal.signal(signal.SIGTERM, shutdown)

健康检查

# Periodic health check in long-running agents
def ensure_connected(client: AgentClient):
    if not client.is_connected():
        logging.warning("Connection lost — reconnecting...")
        client.connect()
    health = client.health_check()
    if not health.ok:
        raise ConnectionError(f"Platform unhealthy: {health.message}")