Building a Multi-Agent System: Orchestrating AI for Maximum Results

Why Multi-Agent Systems?

A single AI agent trying to handle everything is like asking one person to simultaneously do research, write code, execute trades, and monitor risk. They'll do some things well and others poorly.

Multi-agent systems assign specialized roles to different agents, just like a well-run company.

The Multi-Agent Architecture

┌─────────────────────────────────────────────────────┐
│                 Orchestrator Agent                   │
│         (Receives task, delegates to specialists)   │
└──────────┬──────────┬──────────┬────────────────────┘
           │          │          │
    ┌──────▼──┐  ┌────▼────┐  ┌─▼──────────┐
    │Research │  │Strategy │  │Execution   │
    │Agent    │  │Agent    │  │Agent       │
    │         │  │         │  │            │
    │• News   │  │• Signal │  │• Place     │
    │• Data   │  │  gen    │  │  orders    │
    │• Trends │  │• Backtest│  │• Monitor  │
    └──────────┘  └─────────┘  └────────────┘
           │          │          │
    ┌──────▼──────────▼──────────▼──────────┐
    │             Risk Manager Agent         │
    │    (Cross-cutting: monitors all       │
    │    agents' actions for risk limits)   │
    └─────────────────────────────────────┘

Implementation with CrewAI

from crewai import Agent, Task, Crew, Process
from crewai.tools import BaseTool
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o", temperature=0.1)

# Define specialized agents
market_researcher = Agent(
    role="Crypto Market Researcher",
    goal="Gather and synthesize market intelligence for trading decisions",
    backstory="""You are an expert crypto analyst who monitors news, on-chain data, 
    and social sentiment. You provide clear, actionable market summaries.""",
    tools=[news_search_tool, onchain_data_tool],
    llm=llm,
    verbose=True
)

strategy_analyst = Agent(
    role="Quantitative Strategy Analyst", 
    goal="Develop and validate trading strategies based on research",
    backstory="""You are a quant trader who specializes in crypto. You analyze market 
    data and research to identify high-probability trading setups.""",
    tools=[backtesting_tool, price_data_tool],
    llm=llm,
    verbose=True
)

risk_manager = Agent(
    role="Risk Management Specialist",
    goal="Ensure all trades comply with risk parameters and protect capital",
    backstory="""You are a risk manager with expertise in position sizing, 
    drawdown limits, and portfolio protection strategies.""",
    tools=[portfolio_tool, risk_calculator_tool],
    llm=llm,
    verbose=True
)

execution_agent = Agent(
    role="Trade Execution Specialist",
    goal="Execute approved trades efficiently with minimal slippage",
    backstory="""You specialize in optimal trade execution across multiple 
    exchanges, minimizing fees and market impact.""",
    tools=[exchange_tool, order_management_tool],
    llm=llm,
    verbose=True
)

Defining the Workflow

# Task 1: Research
research_task = Task(
    description="""
    Analyze current market conditions for {assets}.
    1. Check latest news and sentiment (last 24 hours)
    2. Review on-chain metrics (whale movements, exchange flows)
    3. Identify key support/resistance levels
    4. Summarize: bullish/bearish thesis, key risks
    """,
    agent=market_researcher,
    expected_output="Structured market analysis with clear directional bias"
)

# Task 2: Strategy (depends on research)
strategy_task = Task(
    description="""
    Based on the market research, identify trading opportunities.
    1. Generate 1-3 specific trade setups
    2. For each: entry, target, stop-loss, position size rationale
    3. Rank by risk/reward ratio
    4. Only include setups with >2:1 R/R
    """,
    agent=strategy_analyst,
    expected_output="Prioritized list of trade setups with parameters",
    context=[research_task]  # Receives research output
)

# Task 3: Risk approval
risk_task = Task(
    description="""
    Review proposed trades against risk parameters:
    - Max 2% portfolio risk per trade
    - Max 10% total portfolio risk at one time
    - No correlated positions > 15%
    Approve, modify, or reject each trade with reasoning.
    """,
    agent=risk_manager,
    expected_output="Approved/rejected trades with adjustments",
    context=[strategy_task]
)

# Task 4: Execution
execution_task = Task(
    description="""
    Execute the risk-approved trades:
    1. Check liquidity on target exchanges
    2. Choose optimal execution venue
    3. Place orders with appropriate limit/market selection
    4. Confirm fills and update position tracker
    """,
    agent=execution_agent,
    expected_output="Execution report with fill prices and confirmations",
    context=[risk_task]
)

# Assemble the crew
trading_crew = Crew(
    agents=[market_researcher, strategy_analyst, risk_manager, execution_agent],
    tasks=[research_task, strategy_task, risk_task, execution_task],
    process=Process.sequential,  # or Process.hierarchical
    verbose=True
)

Running the System

async def run_trading_cycle(assets: list[str]):
    print(f"Starting trading cycle for: {assets}")
    
    result = await trading_crew.kickoff_async(
        inputs={"assets": ", ".join(assets)}
    )
    
    return result

# Run every hour
import asyncio
import schedule

async def hourly_cycle():
    await run_trading_cycle(["BTC", "ETH", "SOL"])

schedule.every().hour.do(lambda: asyncio.run(hourly_cycle()))

LangGraph Alternative

LangGraph gives you more explicit control over agent workflows:

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated

class TradingState(TypedDict):
    assets: list[str]
    research: str | None
    strategies: list[dict] | None
    approved_trades: list[dict] | None
    execution_results: list[dict] | None

def research_node(state: TradingState) -> TradingState:
    # Call market researcher agent
    research = market_researcher.run(state["assets"])
    return {"research": research}

def strategy_node(state: TradingState) -> TradingState:
    strategies = strategy_analyst.run(state["research"])
    return {"strategies": strategies}

def risk_node(state: TradingState) -> TradingState:
    approved = risk_manager.run(state["strategies"])
    return {"approved_trades": approved}

def execution_node(state: TradingState) -> TradingState:
    results = execution_agent.run(state["approved_trades"])
    return {"execution_results": results}

# Build the graph
workflow = StateGraph(TradingState)
workflow.add_node("research", research_node)
workflow.add_node("strategy", strategy_node)
workflow.add_node("risk", risk_node)
workflow.add_node("execution", execution_node)

workflow.set_entry_point("research")
workflow.add_edge("research", "strategy")
workflow.add_edge("strategy", "risk")
workflow.add_edge("risk", "execution")
workflow.add_edge("execution", END)

app = workflow.compile()

Best Practices

1. Keep agents focused — One responsibility per agent
2. Explicit state — Pass data explicitly between agents, don't rely on memory alone
3. Human-in-the-loop — For large trades, require human approval before execution
4. Logging everything — Multi-agent systems are hard to debug; log each agent's inputs and outputs
5. Test each agent independently before testing the full system

Get the Full System

Our multi-agent trading system template is available on the Tools page. It includes all the agents above, plus monitoring, logging, and a web dashboard.