1from openai import OpenAI
2import json
3from typing import Callable
4 
5client = OpenAI()
6 
7# Level 1: Basic tool-use agent with explicit tool definitions
8# The model decides which tool to call and with what arguments
Level 1: Tool-use agent — model decides which tools to call and in what order
9 
10tools = [
11    {
12        "type": "function",
13        "function": {
14            "name": "search_database",
15            "description": "Search the customer database for records matching the query",
16            "parameters": {
17                "type": "object",
18                "properties": {
19                    "query": {"type": "string", "description": "Search query"},
20                    "limit": {"type": "integer", "description": "Max records to return", "default": 10}
21                },
22                "required": ["query"]
23            }
24        }
25    },
26    {
Tool description is critical — "Only call after user confirmation" is a behavioral constraint in the description
27        "type": "function",
28        "function": {
29            "name": "send_email",
30            "description": "Send an email to a customer. Only call after user confirmation.",
31            "parameters": {
32                "type": "object",
33                "properties": {
34                    "to": {"type": "string"},
35                    "subject": {"type": "string"},
36                    "body": {"type": "string"}
37                },
38                "required": ["to", "subject", "body"]
39            }
40        }
41    }
The while loop runs until the model produces a response without tool calls (task complete)
42]
43 
44def run_agent(user_request: str, tool_implementations: dict[str, Callable]) -> str:
45    messages = [{"role": "user", "content": user_request}]
46 
47    while True:
48        response = client.chat.completions.create(
49            model="gpt-4o",
No maximum iteration limit — this agent can loop infinitely if the model never terminates. Add max_turns!
50            messages=messages,
51            tools=tools,
52            tool_choice="auto"
53        )
54 
55        msg = response.choices[0].message
56 
57        if not msg.tool_calls:
58            return msg.content  # Agent is done
59 
60        # Execute each tool call
61        messages.append(msg)
62        for call in msg.tool_calls:
63            tool_name = call.function.name
64            args = json.loads(call.function.arguments)
65            result = tool_implementations[tool_name](**args)
66            messages.append({
67                "role": "tool",
68                "tool_call_id": call.id,
69                "content": json.dumps(result)
70            })

1from openai import OpenAI
2from enum import Enum
3from typing import Callable, Any
4import json
5 
6client = OpenAI()
7 
ActionRisk enum: classify each tool by its reversibility — READ_ONLY, REVERSIBLE, IRREVERSIBLE
8class ActionRisk(Enum):
9    READ_ONLY = "read_only"       # Safe to execute automatically
10    REVERSIBLE = "reversible"     # Confirm with user if uncertain
11    IRREVERSIBLE = "irreversible" # Always confirm with user
12 
13class SafeAgent:
14    """
15    Agent with safety controls:
16    - Maximum iteration limit (prevents infinite loops)
17    - Risk-based human confirmation for irreversible actions
18    - Tool error handling with graceful degradation
19    """
20 
max_turns=10: CRITICAL — prevents infinite loops that consume budget and take endless irreversible actions
21    def __init__(self, max_turns: int = 10):
22        self.max_turns = max_turns
23        self.tool_registry: dict[str, tuple[Callable, ActionRisk]] = {}
24        self.turn_count = 0
25 
26    def register_tool(self, name: str, fn: Callable, risk: ActionRisk, schema: dict):
27        self.tool_registry[name] = (fn, risk)
28 
29    def run(self, user_request: str, require_confirmation: Callable | None = None) -> str:
30        messages = [{"role": "user", "content": user_request}]
31        self.turn_count = 0
32 
33        while self.turn_count < self.max_turns:
Turn count check: agent stops and returns failure message when max_turns is reached
34            self.turn_count += 1
35            response = client.chat.completions.create(
36                model="gpt-4o",
37                messages=messages,
38                tools=self._get_tool_schemas(),
39            )
40 
41            msg = response.choices[0].message
42            if not msg.tool_calls:
43                return msg.content  # Task complete
44 
45            messages.append(msg)
46            for call in msg.tool_calls:
47                tool_name = call.function.name
Confirmation gate for irreversible actions: human-in-the-loop before emails are sent, files deleted, etc.
48                args = json.loads(call.function.arguments)
49 
50                fn, risk = self.tool_registry[tool_name]
51 
52                # Require confirmation for irreversible actions
53                if risk == ActionRisk.IRREVERSIBLE and require_confirmation:
54                    confirmed = require_confirmation(
55                        f"Agent wants to call {tool_name} with args: {args}. Approve?"
56                    )
57                    if not confirmed:
58                        result = {"error": "Action cancelled by user", "args": args}
59                    else:
60                        result = self._safe_execute(fn, args)
_safe_execute: wrap all tool calls in try/except — tool failures should not crash the agent loop
61                else:
62                    result = self._safe_execute(fn, args)
63 
64                messages.append({"role": "tool", "tool_call_id": call.id,
65                                "content": json.dumps(result)})
66 
67        return f"Agent reached maximum turns ({self.max_turns}) without completing task."
68 
69    def _safe_execute(self, fn: Callable, args: dict) -> Any:
70        try:
71            return {"success": True, "result": fn(**args)}
72        except Exception as e:
73            return {"success": False, "error": str(e)}

1from openai import OpenAI
2from anthropic import Anthropic
3 
4openai_client = OpenAI()
5anthropic_client = Anthropic()
6 
7# Level 3: Multi-agent orchestration
8# Orchestrator decomposes task, delegates to specialized sub-agents
9# Each sub-agent has a specific role and limited tool access
10 
Level 3: Multi-agent orchestration — separate agents for planning, research, analysis, and review
11class ResearchOrchestrator:
12    """
13    Orchestrator pattern: one model coordinates multiple specialized agents.
14    Each sub-agent is sandboxed with only the tools it needs.
Orchestrator pattern: one "thinking" model decomposes tasks, specialized agents execute
15    """
16 
17    def run_research_pipeline(self, research_question: str) -> dict:
18        # Step 1: Orchestrator decomposes the task
19        plan = self._plan_research(research_question)
20 
21        # Step 2: Research agent gathers information (READ-ONLY tools only)
Human approval gate before the pipeline produces any output — irreversible actions require explicit confirmation
22        raw_research = self._research_agent(plan["search_queries"])
23 
24        # Step 3: Analysis agent synthesizes findings (no external tools)
25        analysis = self._analysis_agent(research_question, raw_research)
26 
27        # Step 4: Review agent checks for errors (READ-ONLY)
28        review = self._review_agent(analysis)
29 
30        # Step 5: Human approval before any writing/publishing
31        return {"analysis": analysis, "review": review, "status": "awaiting_approval"}
32 
33    def _plan_research(self, question: str) -> dict:
34        """Orchestrator: decompose task into sub-tasks."""
35        response = openai_client.chat.completions.create(
36            model="gpt-4o",
Research agent: READ-ONLY tools only. Cannot write files, send emails, or take side-effecting actions.
37            messages=[{
38                "role": "user",
39                "content": f"Decompose this research question into 3-5 specific search queries: {question}"
40            }],
41            response_format={"type": "json_object"},
42        )
43        return {"search_queries": response.choices[0].message.content}
44 
45    def _research_agent(self, queries: str) -> list[str]:
46        """Research sub-agent: web search + database lookup (READ ONLY)."""
47        results = []
Analysis agent: no external tools at all. Pure reasoning only — reduces attack surface and failure modes.
48        for query in self._parse_queries(queries):
49            results.append(self._web_search(query))  # Read-only tool only
50        return results
51 
52    def _analysis_agent(self, question: str, research: list) -> str:
53        """Analysis sub-agent: synthesize findings. No external tools — only reasoning."""
54        response = anthropic_client.messages.create(
55            model="claude-3-5-sonnet-20241022",
56            max_tokens=4096,
57            messages=[{
58                "role": "user",
59                "content": f"Question: {question}\n\nResearch: {research}\n\nSynthesize findings."
60            }]
61        )
62        return response.content[0].text