并行工具调用：同时等待，按编号收回结果

mkdir -Force "F:\学习\agentdepth-l2-parallel-tools"
cd "F:\学习\agentdepth-l2-parallel-tools"

"""限制并发数量，按调用编号收集固定工具结果。"""

import asyncio
import json
from dataclasses import dataclass
from typing import List


@dataclass(frozen=True)
class Call:
    call_id: str
    tool: str
    argument: str
    delay: float


_RESULTS = {("weather", "上海"): "上海:晴", ("fx", "CNY"): "CNY:1.00"}
_ALLOWED = {"weather", "fx"}


async def run_calls(calls: List[Call], *, limit: int = 2, timeout: float = 0.1) -> dict:
    """调度独立读取，将每项最终结果绑定到它的调用编号。"""
    if type(limit) is not int or limit < 1:
        raise ValueError("invalid_limit")
    if isinstance(timeout, bool) or not isinstance(timeout, (int, float)) or timeout <= 0:
        raise ValueError("invalid_timeout")
    ids = [call.call_id for call in calls]
    if any(not isinstance(call_id, str) or not call_id.strip() for call_id in ids):
        raise ValueError("missing_call_id")
    if len(ids) != len(set(ids)):
        raise ValueError("duplicate_call_id")
    for call in calls:
        if call.tool not in _ALLOWED:
            raise ValueError("tool_not_allowed")
        if not isinstance(call.argument, str) or isinstance(call.delay, bool) or not isinstance(call.delay, (int, float)) or call.delay < 0:
            raise ValueError("invalid_call")

    semaphore = asyncio.Semaphore(limit)
    completion_order = []
    active = 0
    peak = 0

    async def one(call: Call):
        nonlocal active, peak
        async with semaphore:
            active += 1
            peak = max(peak, active)
            try:
                await asyncio.wait_for(asyncio.sleep(call.delay), timeout=timeout)
                value = _RESULTS.get((call.tool, call.argument))
                result = {"status": "ok", "value": value} if value is not None else {
                    "status": "error", "code": "bad_argument"}
            except asyncio.TimeoutError:
                result = {"status": "timeout"}
            finally:
                active -= 1
            completion_order.append(call.call_id)
            return call.call_id, result

    completed = await asyncio.gather(*(one(call) for call in calls))
    return {"completion_order": completion_order,
            "by_call_id": dict(completed), "peak": peak}


if __name__ == "__main__":
    sample = [Call("A", "weather", "上海", 0.03), Call("B", "fx", "CNY", 0.0)]
    print(json.dumps(asyncio.run(run_calls(sample)), ensure_ascii=False, indent=2))

python parallel_lab.py

{
  "completion_order": [
    "B",
    "A"
  ],
  "by_call_id": {
    "A": {
      "status": "ok",
      "value": "上海:晴"
    },
    "B": {
      "status": "ok",
      "value": "CNY:1.00"
    }
  },
  "peak": 2
}

python -m unittest test_parallel_lab -v