Agent skill
multi-agent-system
Design and orchestrate multi-agent AI systems with knowledge harvesting, agent collaboration, and learning loops. Use when working on PSI Engine or similar autonomous agent projects.
Stars
163
Forks
31
Install this agent skill to your Project
npx add-skill https://github.com/majiayu000/claude-skill-registry/tree/main/skills/data/multi-agent-system
SKILL.md
๐ค Multi-Agent System Skill
Use Cases
- Agent spawning & lifecycle management
- Knowledge harvesting from completed tasks
- Agent-to-agent communication
- Learning loop implementation
Agent Architecture
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
โ Orchestrator โ
โ (Assign tasks, monitor, coordinate) โ
โโโโโโโโโโโโโโโโโโฌโโโโโโโโโโโโโโโโโโโโโโโโโ
โ
โโโโโโโโโโโโโโผโโโโโโโโโโโโโ
โผ โผ โผ
โโโโโโโโโโ โโโโโโโโโโ โโโโโโโโโโ
โ Agent 1โ โ Agent 2โ โ Agent 3โ
โ (Task) โ โ (Task) โ โ (Task) โ
โโโโโโฌโโโโ โโโโโโฌโโโโ โโโโโโฌโโโโ
โ โ โ
โโโโโโโโโโโโโดโโโโโโโโโโโโ
โ
โผ
โโโโโโโโโโโโโโโโโโ
โ Knowledge Base โ
โ (ChromaDB) โ
โโโโโโโโโโโโโโโโโโ
Agent Lifecycle
1. Spawn Agent
python
def spawn_agent(agent_id: str, task: str):
# Create PTY for agent terminal
master, slave = pty.openpty()
# Spawn process
process = subprocess.Popen(
['claude', '--task', task],
stdin=slave,
stdout=slave,
stderr=slave,
start_new_session=True
)
return {
'id': agent_id,
'process': process,
'master_fd': master,
'status': 'running'
}
2. Monitor Agent
python
def monitor_agent(agent):
# Read output non-blocking
ready, _, _ = select.select([agent['master_fd']], [], [], 0.1)
if ready:
output = os.read(agent['master_fd'], 4096).decode()
return output
return None
3. Harvest Knowledge
python
def harvest_knowledge(completed_task):
# Extract learnings
learnings = {
'task': completed_task['description'],
'solution': completed_task['output'],
'patterns': extract_patterns(completed_task['output']),
'timestamp': datetime.now().isoformat()
}
# Store in vector DB
collection.add(
documents=[learnings['solution']],
metadatas=[learnings],
ids=[f"learning_{uuid.uuid4()}"]
)
ChromaDB Integration
Setup
python
import chromadb
client = chromadb.Client()
collection = client.get_or_create_collection("knowledge_base")
Store
python
collection.add(
documents=["Solution text here"],
metadatas=[{"source": "agent_1", "task": "debug"}],
ids=["unique_id"]
)
Query (RAG)
python
results = collection.query(
query_texts=["How to fix null pointer?"],
n_results=5
)
Learning Loop
โโโโโโโโโโโโโโโโ
โ Agent runs โ
โ task โ
โโโโโโโโฌโโโโโโโโ
โผ
โโโโโโโโโโโโโโโโ
โ Task result โ
โ extracted โ
โโโโโโโโฌโโโโโโโโ
โผ
โโโโโโโโโโโโโโโโ
โ Knowledge โ โ Store patterns, solutions
โ harvested โ
โโโโโโโโฌโโโโโโโโ
โผ
โโโโโโโโโโโโโโโโ
โ Next agent โ โ Query relevant context
โ uses context โ
โโโโโโโโโโโโโโโโ
Decision Tree
Multi-agent task?
โโโ Need new agent? โ spawn_agent()
โโโ Agent stuck? โ Check PTY buffer, restart if needed
โโโ Task complete? โ Harvest knowledge โ ChromaDB
โโโ Similar task? โ Query ChromaDB for context
โโโ Coordination? โ Use message queue/shared state
Common Issues
| เธเธฑเธเธซเธฒ | เธชเธฒเนเธซเธเธธ | เนเธเนเนเธ |
|---|---|---|
| Agent 3 malfunction | PTY buffer full | Increase buffer / flush regularly |
| Terminal blank | Non-blocking read timing | Use select() with timeout |
| Busy false positive | Status not reset | Reset status after task complete |
| Knowledge not found | Wrong embedding | Tune ChromaDB collection settings |
PSI Engine Specific
- PTY Manager: Always close unused file descriptors
- Agent Status: Use enum (IDLE, RUNNING, COMPLETE, ERROR)
- Harvest timing: Only harvest after verified completion
- Context injection: Limit to 5 most relevant results
Didn't find tool you were looking for?