From Solo to Team: Multi‑Agent Collaboration with AutoGen, CrewAI, and LangGraph

Why Multi‑Agent Collaboration?

Complex real‑world tasks often exceed the capabilities of a single agent because of limited context length, inability to assume multiple professional roles, single‑point failure, and serial execution bottlenecks.

Limitations of a Single Agent

Limited context – cannot handle very long task chains.

Role conflict – hard to act simultaneously as product manager, developer, tester, etc.

Single‑point failure – an error aborts the whole task.

Efficiency bottleneck – tasks run serially and cannot be parallelised.

Advantages of Multiple Agents

Clear division of labour.

Each agent can specialise in a professional capability.

Parallel execution of independent sub‑tasks.

Higher fault tolerance; failure of one agent does not collapse the whole workflow.

Multi‑Agent Architecture Patterns

Master‑Slave : a coordinator agent orchestrates several worker agents; suitable for complex tasks that need clear division of work.

Peer‑to‑Peer : agents cooperate on equal footing; ideal for simple tasks with a shared goal.

Hierarchical : a tree‑like structure where higher‑level agents delegate to lower‑level agents; fits large‑scale projects.

Pipeline (Flow‑Based) : agents pass results along a chain, each handling a stage; works well for assembly‑line style workflows.

AutoGen Hands‑On (Microsoft)

Installation & Configuration

pip install pyautogen

# config.py
import autogen

config_list = [{
    "model": "gpt-4",
    "api_key": "your-api-key",
    "base_url": "https://api.openai.com/v1"
}]

llm_config = {
    "config_list": config_list,
    "temperature": 0.7,
    "timeout": 120,
}

Basic Multi‑Agent System

# basic_multi_agent.py
import autogen

assistant = autogen.AssistantAgent(
    name="assistant",
    llm_config=llm_config,
    system_message="你是一个通用助手，可以帮助解决问题。"
)

user_proxy = autogen.UserProxyAgent(
    name="user_proxy",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=10,
    is_termination_msg=lambda x: x.get("content", "").rstrip().endswith("TERMINATE"),
    code_execution_config={"work_dir": "coding", "use_docker": False},
)

user_proxy.initiate_chat(
    assistant,
    message="帮我写一个 Python 函数，计算斐波那契数列的第 n 项"
)

Programming‑Team Multi‑Agent

# coding_team.py
import autogen

product_manager = autogen.AssistantAgent(
    name="product_manager",
    llm_config=llm_config,
    system_message="你是产品经理，负责分析需求、拆解任务、验收结果。"
)

architect = autogen.AssistantAgent(
    name="architect",
    llm_config=llm_config,
    system_message="你是架构师，负责技术方案设计和模块划分。"
)

developer = autogen.AssistantAgent(
    name="developer",
    llm_config=llm_config,
    system_message="你是开发工程师，负责编写代码实现功能。"
)

tester = autogen.AssistantAgent(
    name="tester",
    llm_config=llm_config,
    system_message="你是测试工程师，负责编写测试用例和验证功能。"
)

user_proxy = autogen.UserProxyAgent(
    name="user_proxy",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=15,
    code_execution_config={"work_dir": "coding", "use_docker": False},
)

groupchat = autogen.GroupChat(
    agents=[user_proxy, product_manager, architect, developer, tester],
    messages=[],
    max_round=20,
)

manager = autogen.GroupChatManager(groupchat=groupchat, llm_config=llm_config)

user_proxy.initiate_chat(
    manager,
    message="""
    我们需要开发一个用户登录功能，包括：
    1. 用户名/密码登录
    2. JWT token 生成和验证
    3. 登录失败限制（5次）
    请产品、架构、开发、测试协作完成。
    """
)

CrewAI Hands‑On (Lightweight)

Installation

pip install crewai

Basic Crew

# basic_crew.py
from crewai import Agent, Task, Crew, Process
from langchain_openai import ChatOpenAI

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

researcher = Agent(
    role="研究员",
    goal="收集和分析市场信息",
    backstory="你是一名资深市场研究员，擅长数据分析和趋势判断",
    verbose=True,
    llm=llm,
)

writer = Agent(
    role="内容写手",
    goal="撰写高质量的技术文章",
    backstory="你是一名技术博主，擅长用通俗语言解释复杂概念",
    verbose=True,
    llm=llm,
)

editor = Agent(
    role="编辑",
    goal="审核和优化文章内容",
    backstory="你是一名资深编辑，对文章质量有极高要求",
    verbose=True,
    llm=llm,
)

research_task = Task(
    description="研究 AI Agent 的最新发展趋势",
    expected_output="一份详细的市场研究报告",
    agent=researcher,
)

writing_task = Task(
    description="基于研究报告，撰写一篇介绍 AI Agent 的文章",
    expected_output="一篇 2000 字左右的技术文章",
    agent=writer,
)

editing_task = Task(
    description="审核并优化文章，确保质量和准确性",
    expected_output="最终发布版本的文章",
    agent=editor,
)

crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[research_task, writing_task, editing_task],
    process=Process.sequential,
    verbose=True,
)

result = crew.kickoff()
print(result)

Custom Tool Integration

# crew_with_tools.py
from crewai import Agent, Task, Crew
from crewai_tools import SerperDevTool, WebsiteSearchTool

search_tool = SerperDevTool()
web_tool = WebsiteSearchTool()

researcher = Agent(
    role="研究员",
    goal="收集 AI 行业最新信息",
    backstory="你擅长使用搜索工具获取实时信息",
    tools=[search_tool, web_tool],
    verbose=True,
    llm=llm,
)

research_task = Task(
    description="搜索 2025 年 AI Agent 领域的最新进展",
    expected_output="一份包含 5 个关键趋势的报告",
    agent=researcher,
)

crew = Crew(agents=[researcher], tasks=[research_task])
result = crew.kickoff()

LangGraph Hands‑On (State‑Machine Multi‑Agent)

Installation

pip install langgraph

State‑Machine Multi‑Agent

# langgraph_multi_agent.py
from typing import TypedDict, List, Annotated
import operator
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI

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

class AgentState(TypedDict):
    messages: Annotated[List[str], operator.add]
    current_agent: str
    result: str

def analyzer(state: AgentState):
    """分析节点"""
    messages = state["messages"]
    response = llm.invoke(f"分析以下需求：{messages[-1]}")
    return {"messages": [response.content], "current_agent": "designer"}

def designer(state: AgentState):
    """设计节点"""
    messages = state["messages"]
    response = llm.invoke(f"基于分析结果设计方案：{messages[-1]}")
    return {"messages": [response.content], "current_agent": "developer"}

def developer(state: AgentState):
    """开发节点"""
    messages = state["messages"]
    response = llm.invoke(f"根据设计方案编写代码：{messages[-1]}")
    return {"messages": [response.content], "current_agent": "tester"}

def tester(state: AgentState):
    """测试节点"""
    messages = state["messages"]
    response = llm.invoke(f"测试以下代码：{messages[-1]}")
    return {"messages": [response.content], "result": response.content, "current_agent": END}

workflow = StateGraph(AgentState)
workflow.add_node("analyzer", analyzer)
workflow.add_node("designer", designer)
workflow.add_node("developer", developer)
workflow.add_node("tester", tester)
workflow.set_entry_point("analyzer")
workflow.add_edge("analyzer", "designer")
workflow.add_edge("designer", "developer")
workflow.add_edge("developer", "tester")
workflow.add_edge("tester", END)

app = workflow.compile()
result = app.invoke({
    "messages": ["开发一个计算器程序，支持加减乘除"],
    "current_agent": "analyzer",
    "result": "",
})
print(result["result"])

Real‑World Example: Intelligent Customer‑Service Team

Team Design Diagram

┌─────────────────────────────────────────────────────────────────┐
│               智能客服团队架构                                 │
├─────────────────────────────────────────────────────────────────┤
│          ┌─────────────┐                                        │
│          │  接待 Agent  │ (接待分流)                              │
│          └──────┬──────┘                                        │
│                 │                                               │
│   ┌────────────────────┼────────────────────┐                │
│   │                    │                    │                │
│   ▼                    ▼                    ▼                │
│ ┌─────────────┐   ┌─────────────┐   ┌─────────────┐          │
│ │ 查询 Agent  │   │ 售后 Agent  │   │ 投诉 Agent  │          │
│ │ (订单/物流) │   │ (退换货)   │   │ (投诉处理) │          │
│ └─────────────┘   └─────────────┘   └─────────────┘          │
│        │                │                │                     │
│        └────────────────────┼────────────────────┘               │
│                         ▼                                   │
│                 ┌─────────────┐                               │
│                 │ 人工 Agent │ (转人工)                         │
│                 └─────────────┘                               │
└─────────────────────────────────────────────────────────────────┘

Full Implementation

# customer_service_team.py
from crewai import Agent, Task, Crew, Process
from langchain_openai import ChatOpenAI
from langchain.tools import tool

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

@tool
def query_order(order_id: str) -> str:
    """查询订单状态"""
    orders = {"1001": "已发货，预计明天送达", "1002": "处理中", "1003": "已签收"}
    return orders.get(order_id, "订单不存在")

@tool
def get_refund_policy() -> str:
    """获取退款政策"""
    return "7天无理由退货，15天质量问题换货"

receptionist = Agent(
    role="接待专员",
    goal="理解用户需求，分流到对应专业 Agent",
    backstory="你是一线客服，负责快速判断用户问题类型并转接",
    verbose=True,
    llm=llm,
)

query_agent = Agent(
    role="查询专员",
    goal="处理订单、物流查询",
    backstory="你是订单查询专家，熟悉订单系统和物流信息",
    tools=[query_order],
    verbose=True,
    llm=llm,
)

after_sales = Agent(
    role="售后专员",
    goal="处理退换货、退款问题",
    backstory="你是售后专家，熟悉退换货流程",
    tools=[get_refund_policy],
    verbose=True,
    llm=llm,
)

complaint_agent = Agent(
    role="投诉专员",
    goal="处理用户投诉，安抚情绪，解决问题",
    backstory="你是投诉处理专家，善于沟通和解决问题",
    verbose=True,
    llm=llm,
)

def create_tasks(user_input: str):
    """根据用户输入创建任务"""
    return [
        Task(
            description=f"分析用户需求并分流：{user_input}",
            expected_output="用户问题类型和应转接的专员",
            agent=receptionist,
        )
    ]

def handle_user_query(user_input: str):
    """处理用户查询"""
    tasks = create_tasks(user_input)
    crew = Crew(
        agents=[receptionist, query_agent, after_sales, complaint_agent],
        tasks=tasks,
        process=Process.sequential,
    )
    return crew.kickoff()

if __name__ == "__main__":
    test_queries = ["查询订单1001", "我想退货", "我要投诉"]
    for q in test_queries:
        print(f"
👤 用户: {q}")
        result = handle_user_query(q)
        print(f"🤖 客服: {result}")

Best Practices

Design Principles

Single responsibility – each agent does one thing.

Clear interfaces – inputs and outputs between agents are well defined.

Fault tolerance – failure of one agent should not affect the whole system.

Observability – record execution logs for every agent.

Common Issues & Solutions

Agent circular calls – set a maximum iteration count.

Context too long – use memory compression techniques.

Task conflicts – introduce a coordinator to arbitrate.

Performance bottlenecks – run independent tasks in parallel.