---

name: llm-agent-orchestration description: > Build and orchestrate LLM-powered agents. Covers agent architectures (ReAct, Plan-and-Execute, multi-agent), tool use and function calling, LangChain agents, LangGraph workflows, CrewAI multi-agent systems, Autogen, agent memory (short-term, long-term, episodic), agent planning and reasoning, error handling and recovery, agent evaluation and testing, human-in-the-loop patterns, agent observability, production agent deployment, and agent safety guardrails. Use when building LLM agents, implementing tool use, orchestrating multi-agent systems, or deploying agents to production. license: Apache-2.0 metadata: author: llmops-skills version: "1.0" category: llmops

LLM Agent Orchestration

Overview

LLM agents extend language models with the ability to reason, plan, use tools, and take actions. Orchestration manages complex multi-step workflows and multi-agent collaboration.

Tested with: langgraph>=1.0, langchain>=1.0, crewai>=1.0, openai>=1.0.

When to Use This Skill

• Building LLM-powered agents with tool use
• Orchestrating multi-step reasoning workflows
• Setting up multi-agent collaboration
• Adding memory to agent conversations
• Deploying agents safely to production

Agent Architecture Patterns

ReAct Agent:                    Plan-and-Execute:
Think → Act → Observe → ...     Plan → Execute Step 1 → ... → Replan

Multi-Agent:                    Hierarchical:
Agent A ←→ Agent B              Supervisor
    ↕           ↕                 ├─ Worker Agent 1
Agent C ←→ Agent D                ├─ Worker Agent 2
                                  └─ Worker Agent 3

Step-by-Step Instructions

1. Tool-Using Agent with OpenAI

from openai import OpenAI
import json

client = OpenAI()

# strict: True + additionalProperties: false enables structured-output
# validation of tool arguments. Under strict mode every property must be listed
# in "required", and JSON Schema "default" is NOT honored - the model either
# supplies a value or you apply the default yourself after parsing.
tools = [
    {
        "type": "function",
        "function": {
            "name": "search_database",
            "description": "Search the product database for items matching a query",
            "strict": True,
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "Search query"},
                    "category": {"type": "string", "enum": ["electronics", "clothing", "books"]},
                    "max_results": {"type": "integer", "description": "Max results, use 5 if unspecified"},
                },
                "required": ["query", "category", "max_results"],
                "additionalProperties": False,
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Get current weather for a location",
            "strict": True,
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {"type": "string"},
                },
                "required": ["location"],
                "additionalProperties": False,
            },
        },
    },
]

def run_agent(user_message, max_iterations=5):
    messages = [
        {"role": "system", "content": "You are a helpful assistant with access to tools."},
        {"role": "user", "content": user_message},
    ]

    for _ in range(max_iterations):
        response = client.chat.completions.create(
            model="gpt-5-mini", messages=messages, tools=tools
        )

        choice = response.choices[0]
        message = choice.message

        if choice.finish_reason == "length":
            return "Response truncated - raise max_tokens or shorten the context"

        if message.tool_calls:
            # Append the assistant turn as a plain dict; passing the SDK object
            # back works today but model_dump() is the documented round-trip.
            messages.append(message.model_dump(exclude_none=True))
            for tool_call in message.tool_calls:
                result = execute_tool(tool_call.function.name,
                                     json.loads(tool_call.function.arguments))
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": json.dumps(result),
                })
        else:
            return message.content

    return "Max iterations reached"

2. LangGraph Workflow

from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
from typing import TypedDict, Annotated
import operator

llm = ChatOpenAI(model="gpt-5-mini")

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    next_action: str

def researcher(state: AgentState):
    """Research step - gather information."""
    messages = state["messages"]
    response = llm.invoke(
        [{"role": "system", "content": "Research the topic thoroughly."}]
        + messages
    )
    return {"messages": [response], "next_action": "analyze"}

def analyzer(state: AgentState):
    """Analysis step - analyze gathered information."""
    response = llm.invoke(
        [{"role": "system", "content": "Analyze the research findings."}]
        + state["messages"]
    )
    return {"messages": [response], "next_action": "write"}

def writer(state: AgentState):
    """Writing step - produce final output."""
    response = llm.invoke(
        [{"role": "system", "content": "Write a clear summary based on the analysis."}]
        + state["messages"]
    )
    return {"messages": [response], "next_action": "end"}

# Build graph. This pipeline is strictly linear, so use plain edges: a
# conditional edge with a single destination adds a routing hop for nothing.
# Reserve add_conditional_edges for real branches.
workflow = StateGraph(AgentState)
workflow.add_node("research", researcher)
workflow.add_node("analyze", analyzer)
workflow.add_node("write", writer)

workflow.add_edge(START, "research")   # modern idiom; set_entry_point still works
workflow.add_edge("research", "analyze")
workflow.add_edge("analyze", "write")
workflow.add_edge("write", END)

app = workflow.compile()
result = app.invoke({"messages": [{"role": "user", "content": "Research MLOps trends"}]})

For a plain tool-calling agent, do not hand-build the graph. LangGraph's langgraph.prebuilt.create_react_agent is deprecated in favour of from langchain.agents import create_agent, which takes a model string, middleware, and a checkpointer:

from langchain.agents import create_agent

agent = create_agent("openai:gpt-5-mini", tools=[search_database, get_weather])
result = agent.invoke({"messages": [{"role": "user", "content": "Weather in Tokyo?"}]})

3. Multi-Agent with CrewAI

from crewai import Agent, Task, Crew, Process

# Define agents
researcher = Agent(
    role="ML Research Analyst",
    goal="Research the latest MLOps tools and best practices",
    backstory="Expert ML engineer with deep knowledge of MLOps ecosystem",
    tools=[search_tool, web_scraper],
    llm="gpt-5-mini",
)

architect = Agent(
    role="ML Systems Architect",
    goal="Design scalable ML pipeline architectures",
    backstory="Senior architect who has designed ML platforms at scale",
    tools=[diagram_tool],
    llm="gpt-5-mini",
)

writer = Agent(
    role="Technical Writer",
    goal="Create clear technical documentation",
    backstory="Experienced technical writer specialized in ML documentation",
    llm="gpt-5-mini",
)

# Define tasks
research_task = Task(
    description="Research the current state of MLOps tooling for {topic}",
    expected_output="Comprehensive research report with tool comparisons",
    agent=researcher,
)

design_task = Task(
    description="Based on the research, design an architecture for {topic}",
    expected_output="Architecture document with diagrams",
    agent=architect,
)

doc_task = Task(
    description="Write user-facing documentation based on the architecture",
    expected_output="Clear, actionable documentation",
    agent=writer,
)

# Run crew
crew = Crew(
    agents=[researcher, architect, writer],
    tasks=[research_task, design_task, doc_task],
    process=Process.sequential,
    verbose=True,
)

result = crew.kickoff(inputs={"topic": "feature store implementation"})

4. Agent Memory

class AgentMemory:
    def __init__(self, max_short_term=20, embeddings_model=None):
        self.short_term = []       # Recent conversation
        self.long_term = []        # Persistent knowledge
        self.max_short_term = max_short_term
        self.embeddings = embeddings_model

    def add_to_short_term(self, message):
        self.short_term.append(message)
        if len(self.short_term) > self.max_short_term:
            # Summarize and move to long-term
            summary = self.summarize(self.short_term[:5])
            self.long_term.append(summary)
            self.short_term = self.short_term[5:]

    def retrieve_relevant(self, query, top_k=3):
        """Retrieve relevant memories for current context."""
        if not self.embeddings or not self.long_term:
            return []
        query_embedding = self.embeddings.encode(query)
        scored = []
        for memory in self.long_term:
            mem_embedding = self.embeddings.encode(memory["content"])
            score = cosine_similarity(query_embedding, mem_embedding)
            scored.append((memory, score))
        scored.sort(key=lambda x: x[1], reverse=True)
        return [m for m, s in scored[:top_k]]

    def build_context(self, current_query):
        """Build full context for the agent."""
        relevant = self.retrieve_relevant(current_query)
        return {
            "short_term": self.short_term[-10:],
            "relevant_memories": relevant,
        }

5. Human-in-the-Loop

class HumanInTheLoopAgent:
    def __init__(self, agent, approval_required_tools=None):
        self.agent = agent
        self.approval_required = approval_required_tools or ["execute_code", "send_email"]

    def run(self, query):
        plan = self.agent.plan(query)

        for step in plan:
            if step.tool in self.approval_required:
                approved = self.request_approval(step)
                if not approved:
                    return "Action cancelled by user"

            result = self.agent.execute_step(step)

            # Check if result needs human review
            if result.confidence < 0.7:
                human_feedback = self.request_review(step, result)
                result = self.agent.refine(result, human_feedback)

        return result

Best Practices

1. Limit tool access - Only give agents the tools they need
2. Never fake a sandbox - eval/exec with a stripped __builtins__ and a blocklist of forbidden substrings is not isolation; both are escapable in one line. Use an AST whitelist for arithmetic (see scripts/build_agent.py) and a container or hosted code-interpreter service for real code execution.
3. Set max iterations - Prevent infinite loops
4. Implement timeouts - Agents can get stuck
5. Log all actions for debugging and audit
6. Use structured outputs for tool arguments (strict: True)
7. Human-in-the-loop for high-stakes actions
8. Test with adversarial inputs - Agents can be manipulated
9. Monitor token usage - Agents can be expensive
10. Use checkpoints for long-running multi-step tasks

Scripts

• scripts/build_agent.py - Agent construction framework
• scripts/multi_agent.py - Multi-agent orchestration setup

References

See references/REFERENCE.md for framework comparisons.

Related skills

Upstream: llm-deployment (the model endpoint agents call — this skill is the application layer on top) · Downstream: llm-observability (agent/tool traces) and llm-guardrails (tool-use safety) See also: llm-prompt-engineering for system prompts and tool descriptions · llm-evaluation for end-to-end agent evals