Multi-Agent Orchestration Without a Framework

1"""A supervisor with two specialists on the OpenAI Python SDK, with no agent framework.
2
3The supervisor is a loop your code owns: the model picks the next tool, the code runs it, and the code decides when to
4stop. Each specialist is a tool backed by its own model call with its own instructions, so control comes back to the
5supervisor after every step.
6"""
7
8import inspect
9import json
10import os
11
12from openai import AzureOpenAI, OpenAI
13
14# On Azure OpenAI, MODEL is your deployment name.
15MODEL = os.environ.get("MODEL", "gpt-4o-mini")
16# A run that needs more than a handful of turns is looping; stop it instead of paying for it.
17MAX_TURNS = 8
18
19
20def make_client():
21    """Azure OpenAI when an endpoint is configured (it reads AZURE_OPENAI_ENDPOINT and AZURE_OPENAI_API_KEY),
22    otherwise the OpenAI API (it reads OPENAI_API_KEY)."""
23    if os.environ.get("AZURE_OPENAI_ENDPOINT"):
24        return AzureOpenAI(api_version=os.environ.get("OPENAI_API_VERSION", "2024-10-21"))
25    return OpenAI()
26
27
28llm = make_client()
29
30
31def search_docs(query: str) -> str:
32    """Search internal documentation and return passages with their source IDs."""
33    # Replace with a real retrieval call. Keep the source IDs so the analysis can cite them.
34    return (
35        "[doc-12] Evaluate retrieval and generation separately. "
36        "[doc-31] Keep a golden set of real questions and run it on every change."
37    )
38
39
40def ask(instructions: str, content: str) -> str:
41    """One model call with its own instructions: all a specialist is, once the framework is gone."""
42    response = llm.chat.completions.create(
43        model=MODEL,
44        messages=[{"role": "system", "content": instructions}, {"role": "user", "content": content}],
45    )
46    return response.choices[0].message.content or ""
47
48
49def research(question: str) -> str:
50    """Find evidence, with source IDs, for a question."""
51    return ask(
52        "You find evidence. From the passages given, return the ones relevant to the question with their source IDs. "
53        "Do not answer the question yourself. If nothing is relevant, say so plainly.",
54        f"Question: {question}\n\nPassages:\n{search_docs(question)}",
55    )
56
57
58def analysis(findings: str) -> str:
59    """Turn research findings into a cited, structured answer."""
60    return ask(
61        "You turn research findings into a short structured answer: at most three points, each citing a source ID. "
62        "Use only the findings you are given. If they do not support an answer, say what is missing.",
63        findings,
64    )
65
66
67SPECIALISTS = {"research": research, "analysis": analysis}
68
69
70def tool(name: str, description: str, argument: str) -> dict:
71    """A tool definition with one required string argument."""
72    return {
73        "type": "function",
74        "function": {
75            "name": name,
76            "description": description,
77            "parameters": {"type": "object", "properties": {argument: {"type": "string"}}, "required": [argument]},
78        },
79    }
80
81
82TOOLS = [
83    tool("research", "Find evidence, with source IDs, for a question.", "question"),
84    tool("analysis", "Turn research findings into a cited, structured answer.", "findings"),
85]
86
87SUPERVISOR = (
88    "You coordinate and never answer from your own knowledge. "
89    "First call research with the user's question. Then call analysis with the findings. "
90    "Return the analysis to the user."
91)
92
93
94def supervise(question: str) -> str:
95    messages = [{"role": "system", "content": SUPERVISOR}, {"role": "user", "content": question}]
96    for _ in range(MAX_TURNS):
97        message = llm.chat.completions.create(model=MODEL, messages=messages, tools=TOOLS).choices[0].message
98        if not message.tool_calls:
99            return message.content or ""
100        messages.append(message.model_dump(exclude_none=True))
101        for call in message.tool_calls:
102            # The code, not the model, decides what runs: unknown tools and malformed arguments are refused.
103            name = call.function.name
104            specialist = SPECIALISTS.get(name)
105            try:
106                args = json.loads(call.function.arguments)
107                inspect.signature(specialist).bind(**args)  # missing or unexpected arguments raise TypeError
108            except (json.JSONDecodeError, TypeError, ValueError):
109                args = None
110            if specialist is None:
111                output = f"Unknown tool: {name}"
112            elif args is None:
113                output = f"Invalid arguments for {name}"
114            else:
115                output = specialist(**args)
116            messages.append({"role": "tool", "tool_call_id": call.id, "content": output})
117    raise RuntimeError(f"Stopped after {MAX_TURNS} turns without an answer.")
118
119
120if __name__ == "__main__":
121    print(supervise("What are the best practices for RAG?"))