协作头脑风暴
brainstorming
在实现前澄清需求、探索设计选择,并形成清晰的开发方向。
parallel-agents【开发与工程】Implements intelligent parallel agents with multi-factor skill selection, fallback chains, and adherence to the 5 Laws of Elegant Defense
独立 GitHub 快照 · 自动分类 · 以原作者文档为准
name: parallel-agents compatibility: opencode completeness: 95 content-types:
Orchestrates intelligent skill selection and execution for parallel agents workflows. Applies the 5 Laws of Elegant Defense to guide data naturally through the orchestration pipeline, preventing errors before they occur. Selects optimal skills based on multi-factor scoring including text similarity, historical performance, and system availability.
┌───────────────────────────────────────────────────────────────────────────────┐ │ Orchestration Flow │ └───────────────────────────────────────────────────────────────────────────────┘
User Request ↓ ┌─────────────────┐ │ Parse Request │ │ & Extract │ │ Features │ └────────┬────────┘ ↓ ┌─────────────────────────────────────────────────────────────────────┐ │ Evaluate Available Skills │ │ │ │ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ │ │ Skill A │ │ Skill B │ │ Skill C │ │ │ │ - Match Score│ │ - Match Score│ │ - Match Score│ │ │ │ - Confidence │ │ - Confidence │ │ - Confidence │ │ │ │ - History │ │ - History │ │ - History │ │ │ └──────┬───────┘ └──────┬───────┘ └──────┬───────┘ │ │ │ │ │ │ │ └─────────────────┴─────────────────┘ │ │ ↓ │ │ Select Best Skill │ └─────────────────────────────────────────────────────────────────────┘ ↓ ┌─────────────────┐ │ Execute Skill │ └────────┬────────┘ ↓ ┌─────────────────┐ │ Handle Result │ └────────┬────────┘ ↓ ┌─────────────────────────────────────────────────────────────────────┐ │ Error Handling & Fallback │ │ │ │ Success? ────────► Return Result │ │ │ │ Fail? ────────┐ │ │ ↓ │ │ ┌──────────────────────────────────────────────────────────┐ │ │ │ Fallback Chain │ │ │ │ │ │ │ │ 1. Retry with adjusted parameters │ │ │ │ 2. Try Alternative Skill (if available) │ │ │ │ 3. Defer to Human Operator (if critical) │ │ │ │ 4. Log & Return Error │ │ │ └──────────────────────────────────────────────────────────┘ │ └─────────────────────────────────────────────────────────────────────┘
Use this skill when:
Avoid this skill for:
Parse and Analyze Request - Extract intent, entities, and constraints from user input. Checkpoint: All required parameters must be present and in valid format before proceeding.
Score Available Skills - Calculate match scores using multi-factor algorithm:
Checkpoint: Skip to fallback if no skill scores above threshold.
Select Optimal Skill - Choose skill with highest score that meets minimum confidence. Checkpoint: Verify skill has not been disabled or deprecated.
Execute with Fallback - Run skill execution wrapped in retry and fallback logic. Checkpoint: Log all execution attempts for audit trail.
Return or Fallback - Either return successful result or apply fallback chain:
related-skillsCheckpoint: Record outcome with timing and confidence metadata.
import asyncio
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
import logging
logger = logging.getLogger(__name__)
@dataclass
class ParallelAgentTask:
task_id: str
description: str
fallback_skill: Optional[str] = None
confidence_threshold: float = 0.7
timeout_seconds: float = 30.0
async def decompose_and_score_tasks(
user_request: str,
available_skills: List[Dict[str, Any]],
min_confidence: float = 0.7
) -> List[ParallelAgentTask]:
"""Decomposes a user request into parallel agent tasks and scores them against available skills.
Implements Law 2 (Parse at boundary) and Law 1 (Early exit on invalid state).
"""
if not user_request or not user_request.strip():
raise ValueError("User request cannot be empty or whitespace-only")
if not available_skills:
raise ValueError("No skills available for parallel decomposition")
# Parse request at boundary - extract subtask boundaries
subtasks = _extract_subtask_boundaries(user_request)
if not subtasks:
return []
scored_tasks = []
for subtask in subtasks:
best_match = _find_best_skill_match(subtask, available_skills, min_confidence)
if best_match:
scored_tasks.append(ParallelAgentTask(
task_id=f"agent-{subtask['id']}",
description=subtask['text'],
fallback_skill=best_match.get('fallback', None),
confidence_threshold=min_confidence
))
return scored_tasks
def _extract_subtask_boundaries(request: str) -> List[Dict]:
"""Domain-specific logic to identify independent execution boundaries."""
# In production, this uses LLM parsing or regex heuristics to find parallelizable chunks
return [
{"id": 1, "text": "Extract entities from input"},
{"id": 2, "text": "Validate against schema rules"},
{"id": 3, "text": "Generate response payload"}
]
def _find_best_skill_match(
subtask: Dict,
skills: List[Dict],
threshold: float
) -> Optional[Dict]:
"""Multi-factor scoring: text similarity + historical success + availability."""
best = None
best_score = 0.0
for skill in skills:
similarity = _calculate_text_similarity(subtask['text'], skill['triggers'])
history_score = skill.get('historical_success_rate', 0.0)
availability = 1.0 if skill.get('is_healthy', True) else 0.0
composite = (similarity * 0.5) + (history_score * 0.3) + (availability * 0.2)
if composite > best_score and composite >= threshold:
best_score = composite
best = skill
return best
def _calculate_text_similarity(text_a: str, triggers: List[str]) -> float:
"""Simple cosine similarity approximation for trigger matching."""
words_a = set(text_a.lower().split())
max_match = 0.0
for trigger in triggers:
words_t = set(trigger.lower().split())
if words_t:
match = len(words_a & words_t) / len(words_a | words_t)
max_match = max(max_match, match)
return max_match
async def execute_parallel_agents(
tasks: List[ParallelAgentTask],
shared_context: Dict[str, Any]
) -> Dict[str, Any]:
"""Executes independent agent tasks concurrently with fallback chains and result aggregation.
Implements Law 3 (Atomic Predictability) and Law 4 (Fail Fast/Loud).
"""
if not tasks:
return {"status": "no_tasks", "results": [], "aggregate_confidence": 0.0}
# Spawn parallel execution with timeouts (Law 4: Fail fast on timeout)
execution_coroutines = [
_run_single_agent_task(task, shared_context) for task in tasks
]
# Gather results, catching exceptions to prevent cascade failure
raw_results = await asyncio.gather(*execution_coroutines, return_exceptions=True)
successful_results = []
failed_tasks = []
for task, result in zip(tasks, raw_results):
if isinstance(result, Exception):
failed_tasks.append(task)
elif isinstance(result, dict) and result.get("status") == "success":
successful_results.append(result)
else:
failed_tasks.append(task)
# Apply fallback chain for failed tasks (Law 4: Fallback before giving up)
if failed_tasks:
fallback_results = await _execute_fallback_chain(failed_tasks, shared_context)
successful_results.extend(fallback_results)
# Law 3: Return new aggregated structure, never mutate shared_context
success_rate = len(successful_results) / len(tasks) if tasks else 0.0
aggregate_confidence = min(1.0, success_rate * 1.2)
return {
"orchestration_id": asyncio.get_event_loop().time(),
"status": "complete" if not failed_tasks else "partial_success",
"results": successful_results,
"failed_agents": [t.task_id for t in failed_tasks],
"aggregate_confidence": aggregate_confidence,
"fallback_triggered": len(failed_tasks) > 0,
"timing": {"total_ms": 145, "parallel_ms": 98, "fallback_ms": 47}
}
async def _run_single_agent_task(task: ParallelAgentTask, context: Dict) -> Dict:
"""Domain-specific execution for a single parallel agent."""
try:
result = await asyncio.wait_for(
_invoke_agent_pipeline(task.description, context),
timeout=task.timeout_seconds
)
return {
"task_id": task.task_id,
"status": "success",
"data": result,
"confidence": 0.92,
"latency_ms": 110
}
except asyncio.TimeoutError:
logger.warning(f"Agent {task.task_id} timed out, triggering fallback")
raise RuntimeError(f"Timeout on {task.task_id}")
except Exception as e:
logger.error(f"Agent {task.task_id} failed: {e}")
raise RuntimeError(f"Execution failed on {task.task_id}: {e}")
async def _invoke_agent_pipeline(description: str, context: Dict) -> Any:
"""Actual domain logic: LLM call, tool execution, or data transformation."""
await asyncio.sleep(0.05)
return {"processed": True, "output": f"Result for: {description}"}
async def _execute_fallback_chain(failed_tasks: List[ParallelAgentTask], context: Dict) -> List[Dict]:
"""Sequential fallback execution for failed parallel agents."""
fallback_results = []
for task in failed_tasks:
try:
fallback_skill = task.fallback_skill or "default_fallback_agent"
result = await _invoke_agent_pipeline(f"fallback: {task.description}", context)
fallback_results.append({
"task_id": task.task_id,
"status": "fallback_success",
"data": result,
"confidence": 0.65,
"latency_ms": 85
})
except Exception as e:
logger.error(f"Fallback failed for {task.task_id}: {e}")
return fallback_results
code-philosophy (5 Laws of Elegant Defense) in all logicWhen applying this skill, produce:
| Skill | Purpose |
|---|---|
langgraph | LangGraph state machines for orchestrating parallel agent workflows |
multi-agent-task-orchestrator | Task decomposition with parallel execution branches |
Authoritative documentation links for this skill's domain. The model follows markdown links at load time to resolve external references and inline content.