Building Autonomous Agent Hierarchies

"""
Minimal orchestrator + sub-agent example against a self-hosted,
OpenAI-compatible endpoint (Ollama, vLLM, or TGI).

Install:
    pip install openai

Run a local model first, e.g.:
    ollama run llama3.1:8b
"""

from openai import OpenAI
import json

# 1. Connect — point the client at your self-hosted server
client = OpenAI(
    base_url="http://localhost:11434/v1",  # Ollama's OpenAI-compatible endpoint
    api_key="not-needed",                   # self-hosted servers usually ignore this
)

MODEL = "llama3.1:8b"


# 2. Define sub-agents — small, focused functions with their own prompts
def research_agent(query: str) -> str:
    """Sub-agent specialised in summarising information for a query."""
    response = client.chat.completions.create(
        model=MODEL,
        messages=[
            {"role": "system", "content": (
                "You are a research specialist. Given a topic, return a "
                "concise, factual summary in 3 bullet points. Do not answer "
                "anything outside the research task."
            )},
            {"role": "user", "content": query},
        ],
        temperature=0.2,
    )
    return response.choices[0].message.content


def coding_agent(spec: str) -> str:
    """Sub-agent specialised in writing small code snippets."""
    response = client.chat.completions.create(
        model=MODEL,
        messages=[
            {"role": "system", "content": (
                "You are a coding specialist. Given a specification, return "
                "only working Python code with no explanation."
            )},
            {"role": "user", "content": spec},
        ],
        temperature=0.1,
    )
    return response.choices[0].message.content


# 3. Build a tool directory — how the orchestrator "sees" its sub-agents
SUBAGENT_TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "research_agent",
            "description": "Use for gathering or summarising factual information.",
            "parameters": {
                "type": "object",
                "properties": {"query": {"type": "string"}},
                "required": ["query"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "coding_agent",
            "description": "Use for writing or fixing small code snippets.",
            "parameters": {
                "type": "object",
                "properties": {"spec": {"type": "string"}},
                "required": ["spec"],
            },
        },
    },
]

DISPATCH = {
    "research_agent": research_agent,
    "coding_agent": coding_agent,
}


# 4. Orchestrator loop — delegate, collect results, repeat until done
def orchestrator(task: str) -> str:
    messages = [
        {"role": "system", "content": (
            "You are the orchestrator. You do not answer tasks yourself — "
            "you decide which specialist tool to call for each part of the "
            "task, then combine their outputs into a final answer."
        )},
        {"role": "user", "content": task},
    ]

    response = client.chat.completions.create(
        model=MODEL,
        messages=messages,
        tools=SUBAGENT_TOOLS,
        tool_choice="auto",
    )
    msg = response.choices[0].message

    while msg.tool_calls:
        messages.append(msg)
        for call in msg.tool_calls:
            fn = DISPATCH[call.function.name]
            args = json.loads(call.function.arguments)
            result = fn(**args)
            messages.append({
                "role": "tool",
                "tool_call_id": call.id,
                "content": result,
            })

        response = client.chat.completions.create(
            model=MODEL,
            messages=messages,
            tools=SUBAGENT_TOOLS,
            tool_choice="auto",
        )
        msg = response.choices[0].message

    return msg.content


# 5. Call it
if __name__ == "__main__":
    result = orchestrator(
        "Summarise what a vector database is, then write a Python "
        "function that computes cosine similarity between two vectors."
    )
    print(result)

$ python orchestrator.py

A vector database stores data as high-dimensional vectors and supports
similarity search rather than exact matching.
- It indexes embeddings so you can retrieve the items "nearest" to a query.
- Common uses include semantic search, recommendation systems, and RAG.
- Popular options include Qdrant, Weaviate, and Chroma.

def cosine_similarity(a, b):
    dot = sum(x * y for x, y in zip(a, b))
    norm_a = sum(x ** 2 for x in a) ** 0.5
    norm_b = sum(y ** 2 for y in b) ** 0.5
    return dot / (norm_a * norm_b)

# ~/.hermes/config.yaml
model:
  default: qwen2.5-coder:32b
  provider: custom
  base_url: http://localhost:11434/v1   # Ollama; use :8000/v1 for vLLM
  api_key: ""                           # empty for local endpoints

# Optional: route sub-agents to a cheaper/faster model
delegation:
  model: "qwen2.5-coder:7b"
  max_iterations: 50        # per-child step limit
  max_concurrent_children: 3
  max_spawn_depth: 1        # raise to 2+ for nested orchestration