TRAVEL ASSISTANT

Agents at the same priority run through asyncio.gather; exceptions are returned as partial failures instead of crashing aggregation.

async def _execute_parallel_agents(
    self, tasks, context, previous_results
):
    parallel_coroutines = []
    for task in tasks:
        agent_name = task.get("agent_name")
        priority = task.get("priority", 0)
        coroutine = self._execute_agent(
            agent_name=agent_name,
            context=context,
            reason=task.get("reason", ""),
            expected_output=task.get("expected_output", ""),
            previous_results=previous_results,
        )
        parallel_coroutines.append((agent_name, priority, coroutine))

    execution_results = await asyncio.gather(
        *[coro for _, _, coro in parallel_coroutines],
        return_exceptions=True,
    )

    results = []
    for (agent_name, priority, _), exec_result in zip(
        parallel_coroutines, execution_results
    ):
        # Exceptions are converted to structured partial failures here.
        results.append({
            "agent_name": agent_name,
            "priority": priority,
            "result": exec_result,
        })
    return results

The registry discovers plugin agents, resolves legacy names, imports the matching module and caches the instance after first use.

def __getitem__(self, agent_name: str):
    if agent_name in self.cache:
        return self.cache[agent_name]

    skill_name = self._resolve_agent_name(agent_name)
    script_path = self._skill_map[skill_name]
    module_name = f"skills.{skill_name}.agent"
    spec = importlib.util.spec_from_file_location(module_name, script_path)
    module = importlib.util.module_from_spec(spec)
    sys.modules[module_name] = module
    spec.loader.exec_module(module)

    for name, obj in inspect.getmembers(module):
        if inspect.isclass(obj) and issubclass(obj, AgentBase):
            agent_class = obj
            break

    agent_instance = agent_class(**init_params)
    self.cache[agent_name] = agent_instance
    return agent_instance

Repeated failures stop further model calls for a recovery window, then allow controlled probing instead of cascading failure.

class CircuitBreaker:
    def record_failure(self) -> None:
        if self._state == CircuitState.HALF_OPEN:
            self._state = CircuitState.OPEN
            self._opened_at = time.monotonic()
            return

        self._failure_count += 1
        if self._failure_count >= self.failure_threshold:
            self._state = CircuitState.OPEN
            self._opened_at = time.monotonic()

    def raise_if_open(self) -> None:
        if not self.allow_call():
            raise CircuitOpenError("服务暂时不可用，请稍后再试")