Multi-Agent AI Systems — Complete Guide: Patterns, Frameworks, and Production Deployment

from langchain_anthropic import ChatAnthropic
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

# Define shared state passed between all agents
class AgentState(TypedDict):
    task: str
    research: str
    draft: str
    review: str
    final: str
    messages: Annotated[list, operator.add]

llm = ChatAnthropic(model="claude-sonnet-4-6")

def research_agent(state: AgentState):
    """Agent 1: Research the topic"""
    response = llm.invoke([
        {"role": "system", "content": "You are a research expert. Gather key facts and cite sources."},
        {"role": "user", "content": f"Research thoroughly: {state['task']}"}
    ])
    return {"research": response.content, "messages": [response]}

def writer_agent(state: AgentState):
    """Agent 2: Write based on research"""
    response = llm.invoke([
        {"role": "system", "content": "You are a professional writer. Be clear and structured."},
        {"role": "user", "content": f"Write about: {state['task']}\n\nResearch: {state['research']}"}
    ])
    return {"draft": response.content}

def reviewer_agent(state: AgentState):
    """Agent 3: Review and provide specific feedback"""
    response = llm.invoke([
        {"role": "system", "content": "You are a critical editor. Be specific about improvements."},
        {"role": "user", "content": f"Review this draft:\n\n{state['draft']}\n\nList specific improvements needed."}
    ])
    return {"review": response.content}

def finalize_agent(state: AgentState):
    """Agent 4: Incorporate review feedback"""
    response = llm.invoke([
        {"role": "user", "content": f"Revise based on review:\n\nDraft: {state['draft']}\n\nReview: {state['review']}"}
    ])
    return {"final": response.content}

# Build the graph
workflow = StateGraph(AgentState)
workflow.add_node("research", research_agent)
workflow.add_node("writer", writer_agent)
workflow.add_node("reviewer", reviewer_agent)
workflow.add_node("finalize", finalize_agent)

# Sequential pipeline
workflow.set_entry_point("research")
workflow.add_edge("research", "writer")
workflow.add_edge("writer", "reviewer")
workflow.add_edge("reviewer", "finalize")
workflow.add_edge("finalize", END)

app = workflow.compile()
result = app.invoke({"task": "Explain quantum computing for developers", "messages": []})
print(result["final"])

import anthropic
import json

client = anthropic.Anthropic()

# Define tools the agent can call
tools = [
    {
        "name": "web_search",
        "description": "Search the web for current information",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "Search query"}
            },
            "required": ["query"]
        }
    },
    {
        "name": "get_database_record",
        "description": "Fetch a record from the product database",
        "input_schema": {
            "type": "object",
            "properties": {
                "product_id": {"type": "string"}
            },
            "required": ["product_id"]
        }
    }
]

def execute_tool(name: str, inputs: dict) -> str:
    """Execute the tool and return result as string."""
    if name == "web_search":
        # Real implementation would call a search API
        return f"Search results for '{inputs['query']}': [placeholder results]"
    elif name == "get_database_record":
        return json.dumps({"id": inputs["product_id"], "name": "Widget", "price": 29.99})

def run_agent(user_message: str) -> str:
    messages = [{"role": "user", "content": user_message}]

    while True:
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=4096,
            tools=tools,
            messages=messages,
        )

        # No tool calls — final answer
        if response.stop_reason == "end_turn":
            return response.content[0].text

        # Process tool calls
        tool_results = []
        for block in response.content:
            if block.type == "tool_use":
                result = execute_tool(block.name, block.input)
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": result,
                })

        # Add agent response + tool results to conversation
        messages.append({"role": "assistant", "content": response.content})
        messages.append({"role": "user", "content": tool_results})

result = run_agent("What are the current reviews for product ID P123?")
print(result)