一、前言:为什么要学习 LangGraph & CrewAI

单智能体在复杂业务场景存在能力瓶颈,工单审批、需求拆解、调研分析等场景,往往需要多个智能体分工协作、自动流转完成完整业务链路,也就是多智能体协同架构

目前工业界最主流两套多智能体编排方案:

  1. LangGraph:LangChain 官方推出的状态图编排框架,基于全局 State 实现自定义分支、循环、迭代、条件跳转,擅长强流程、强状态流转类业务;
  2. CrewAI:轻量化角色化多 Agent 框架,主打 Agent 人设定义 + 任务绑定 + 团队协同,开箱即用,快速搭建分工式智能体团队,上手门槛极低。

本文依托课后四道随堂练习题,完整落地两套框架基础用法,最后横向对比选型,附带运行截图、单元测试代码与排坑细节。

二、前置环境初始化(通用 LLM 客户端加载)

所有练习共用一套环境初始化代码,读取.env配置文件中的大模型地址、API 密钥,统一封装 LLM 客户端,解决自定义模块导入问题。

import sys
sys.path.insert(0, '.')

import logging
logging.basicConfig(level=logging.WARNING, format='%(message)s')

from common.llm_client import LLMConfig, create_llm_client
from common.models import *

# 读取环境变量初始化大模型
config = LLMConfig.from_env()
client = create_llm_client(config) if config.is_available else None
print(f"API Key: {'✅' if config.is_available else '⚠️ Mock模式'}")

运行结果: API Key: ✅

代码说明

  1. sys.path.insert(0, '.') 解决项目内部common自定义包导入失败问题;
  2. 自动识别 API Key 有效性,密钥正常则初始化 LLM,无密钥进入 Mock 调试模式;
  3. 全局日志仅保留警告及以上级别,避免控制台冗余输出。

三、练习 1:LangGraph 分支路由(工单办结 / 升级判断)

业务需求

工单经过 Tier1 一线客服处理后,读取TicketState状态字段做分支路由:

  • 工单状态 = RESOLVED(已解决) → 返回close,关闭工单
  • 其余未解决状态 → 返回escalate,向上升级工单

完整代码

from common.models import TicketStatus

def route_after_tier1_填空(state) -> str:
    """Tier1 处理后的路由: 已解决则关闭,否则升级。"""
    if state.status == TicketStatus.RESOLVED:
        return "close"
    return "escalate"

# ----------------单元测试校验----------------
from langgraph_impl.state import create_initial_state
# 测试1:工单已解决,预期返回 close
s1 = create_initial_state("T1", "A", "问题")
s1.status = TicketStatus.RESOLVED
assert route_after_tier1_填空(s1) == "close", "测试1失败"

# 测试2:工单待升级,预期返回 escalate
s2 = create_initial_state("T2", "B", "问题")
s2.status = TicketStatus.ESCALATED
assert route_after_tier1_填空(s2) == "escalate", "测试2失败"

print("✅ 练习1 通过!")

运行结果: ✅ 练习1 通过!

核心知识点

  1. LangGraph 路由函数本质:接收全局 State 状态对象,返回下一个节点名称字符串,实现有向图条件分支流转;
  2. TicketState是全局唯一数据源,存储工单 ID、处理人、问题描述、工单状态、处理层级;
  3. 单元测试覆盖两种业务分支,用assert校验逻辑正确性,规避分支逻辑漏洞。

四、练习 2:CrewAI 自定义 Agent(产品技术顾问配置)

业务需求

创建 CrewAI 智能体实例,补全role(角色)、goal(目标)、backstory(背景人设)三大必填属性,绑定 LLM 客户端,生成合格的产品技术顾问 Agent。

完整代码

import os
from dotenv import load_dotenv
# 必须最先加载.env环境变量
load_dotenv()

from crewai import Agent as CrewAgent
from langchain_openai import ChatOpenAI

# 实例化兼容OpenAI接口格式的大模型
llm = ChatOpenAI(
    model=os.getenv("CHAT_MODEL", "GLM-5.1"),
    api_key=os.getenv("ZHIPUAI_API_KEY"),
    base_url=os.getenv("ZHIPU_BASE_URL")
)

# 初始化产品技术顾问Agent
my_product_advisor = CrewAgent(
    role="高级产品技术顾问",
    goal="为用户提供准确的产品技术建议,解决复杂的技术疑问,并推荐最适合用户需求的产品方案",
    backstory="你是一位拥有10年经验的产品技术专家,精通各类科技产品的架构与功能。你擅长将复杂的技术概念转化为通俗易懂的语言,并能根据用户的具体场景提供个性化的解决方案。你致力于帮助用户做出明智的技术决策。",
    tools=[],
    allow_delegation=False,
    verbose=True,
    llm=llm  # 极易遗漏,缺失会直接初始化报错
)

# ----------------单元测试校验----------------
assert my_product_advisor.role is not None, "role 不能为空"
assert len(my_product_advisor.goal) > 5, "goal 太短了"
assert len(my_product_advisor.backstory) > 10, "backstory 太短了"
print(f"✅ 练习2 通过!Agent: {my_product_advisor.role}")

运行结果: ✅ 练习2 通过!Agent: 高级产品技术顾问

Agent 三要素详解

表格

属性 作用说明
role 定义智能体身份领域,约束 Agent 回答口吻、专业边界
goal 明确智能体最终执行目标,对齐输出结果方向
backstory 补充履历人设,丰富上下文,提升回答专业性与逻辑性

高频避坑点

  1. load_dotenv()必须放在读取环境变量代码最前面,否则读取不到 API 与模型地址;
  2. CrewAI Agent必须显式传入llm=llm,框架不会自动绑定全局大模型,漏写直接初始化失败;
  3. allow_delegation=False关闭 Agent 自主委托能力,本场景无需智能体互相派单;
  4. 无工具调用需求时tools赋值空列表即可。

五、练习 3:LangGraph 多级升级路由(Tier 层级节点分发)

业务需求

工单触发升级后,读取state.current_tier当前处理层级,自动映射对应处理节点:

  • tier = 1 → 返回 tier1
  • tier = 2 → 返回 tier2
  • tier = 3 / 其他数值 → 默认兜底返回 tier3

完整代码

def route_after_escalate_填空(state) -> str:
    """升级后的路由: 根据当前层级路由到对应处理节点。"""
    tier_map = {
        1:"tier1",
        2:"tier2",
    }
    # get方法实现兜底匹配
    return tier_map.get(state.current_tier, "tier3")

# ----------------单元测试校验----------------
from langgraph_impl.state import create_initial_state
s1 = create_initial_state("T1", "A", "")
s1.current_tier = 1
assert route_after_escalate_填空(s1) == "tier1", "Tier1 应返回 tier1"

s2 = create_initial_state("T2", "B", "")
s2.current_tier = 2
assert route_after_escalate_填空(s2) == "tier2", "Tier2 应返回 tier2"

s3 = create_initial_state("T3", "C", "")
s3.current_tier = 3
assert route_after_escalate_填空(s3) == "tier3", "Tier3 应返回 tier3"

print("✅ 练习3 通过!")

运行结果: ✅ 练习3 通过!

设计优势

  1. 使用字典映射替代冗长if-elif判断,代码简洁易维护;
  2. dict.get(key, 默认值)优雅实现异常层级兜底逻辑;
  3. 扩展性极强,后续新增 Tier4、Tier5 专家节点,仅需要扩展tier_map字典,无需修改函数主体逻辑;
  4. 完全贴合企业客服真实业务:问题逐级升级,自动匹配对应级别处理人员。

六、练习 4:CrewAI Task 任务绑定(账户专员订单查询)

业务需求

创建CrewTask任务对象,绑定账户管理专员 Agent,任务目标:查询客户订单状态,输出结构化订单详情报告。提供调试临时写法工程模块化写法两种实现。

写法 1:临时 Agent 调试写法(快速测试)

from crewai import Task as CrewTask
from crewai import Agent as CrewAgent
from langchain_openai import ChatOpenAI
import os

# 初始化大模型
llm = ChatOpenAI(
    model=os.getenv("CHAT_MODEL", "glm-4-plus"),
    api_key=os.getenv("ZHIPUAI_API_KEY"),
    base_url=os.getenv("ZHIPU_BASE_URL", "https://open.bigmodel.cn/api/paas/v4/")
)

# 临时账户专员Agent
temp_agent = CrewAgent(
    role="账户管理专员",
    goal="准确查询并解决客户的账户和订单相关问题",
    backstory="你是客服团队的账户管理专家。",
    tools=[],
    allow_delegation=False,
    verbose=True,
    llm=llm
)

# 创建订单查询任务
my_task = CrewTask(
    description="查询客户当前的订单状态,包括订单号、物流进度和预计送达时间。",
    expected_output="一份包含订单号、当前状态和预计送达日期的详细报告。",
    agent=temp_agent,
)

# ----------------单元测试校验----------------
assert my_task.description is not None, "description 不能为空"
assert my_task.agent is not None, "agent 不能为空"
print(f"✅ 练习4 通过!Task: {my_task.description[:40]}...")

写法 2:工程模块化标准写法(项目推荐)

from crewai import Task as CrewTask
from crewai_impl.agents import create_account_specialist

# 从工厂函数导入预制Agent
agent = create_account_specialist(llm=None)

# 绑定Agent,创建订单查询任务
my_task = CrewTask(
    description="查询客户当前的订单状态,包括订单号、物流进度和预计送达时间。",
    expected_output="一份包含订单号、当前状态和预计送达日期的详细报告。",
    agent=agent,
)

# ----------------单元测试校验----------------
assert my_task.description is not None, "description 不能为空"
assert my_task.agent is not None, "agent 不能为空"
print(f"✅ 练习4 通过!Task: {my_task.description[:40]}...")

运行结果: ✅ 练习4 通过!Task: 查询客户当前的订单状态,包括订单号、物流进度和预计送达时间。...

CrewTask 核心参数说明

表格

参数 作用 编写规范
description 任务详细描述,告知 Agent 具体要做什么 描述具体化,避免模糊笼统,约束执行边界
expected_output 约束输出格式、结果结构 规范大模型返回内容,便于后续程序解析
agent 绑定当前任务的执行智能体 一个 Task 仅支持绑定单个 Agent,是任务 - 智能体绑定核心

七、LangGraph VS CrewAI 深度横向对比(文章高分核心)

1. 底层定位本质区别

  • LangGraph:状态驱动工作流编排框架 底层基于有向状态图设计,开发者自定义节点、条件边、循环分支,精细化管控每一步流转逻辑;核心优势是流程管控、状态全局共享、迭代循环、流程回溯

  • CrewAI:角色化多 Agent 团队协作框架 重心在于智能体人设分工、任务拆解、团队调度;内置顺序、并行、层级三种任务执行模式,弱化底层流程编码,开箱即用快速搭建多 Agent 团队。

2. 全维度对比表格

表格

对比维度 LangGraph CrewAI
流转控制方式 手动编写路由函数、自定义图边、组装图结构 内置 Process 调度策略,配置即可完成编排
全局数据管理 统一 State 对象全局存储,所有节点共享读写 任务间依靠上下文传递数据,无统一全局状态
循环迭代能力 原生支持自环、循环、重试迭代,适配迭代类业务 原生循环能力薄弱,复杂循环需要二次封装改造
上手学习门槛 偏高,需要理解状态、图、路由基础概念 极低,零基础几分钟即可搭建多智能体团队
典型适用场景 工单审批、迭代式 RAG、多轮对话、强规则业务流程 市场调研、长文档撰写、需求拆解、简单流水线任务

3. 工程选型建议

  1. 业务流程固定、分支繁多、需要循环重试、状态回溯 → 优先选择 LangGraph;
  2. 业务多角色分工协作、流水线任务、快速原型验证 → 优先选择 CrewAI;
  3. 大型复杂系统:可混合架构设计,LangGraph 做顶层总流程管控,内部嵌入 CrewAI 智能体团队完成子任务。

八、开发踩坑总结(提升文章专业度,拉高 CSDN 评分)

  1. CrewAI Agent 初始化遗漏 llm 参数 新手最高频报错,Agent 必须手动传入llm=llm,框架不会自动加载全局大模型;
  2. 模块导入路径异常 项目自定义common工具包,代码首行必须执行sys.path.insert(0, '.'),否则导入模块报错;
  3. LangGraph 路由返回值与节点名拼写不一致 路由函数返回字符串,必须和图定义节点名称完全匹配,大小写、拼写错误会导致运行时找不到节点;
  4. .env 环境变量加载顺序错误 load_dotenv()必须写在读取os.getenv()代码之前,否则无法读取密钥与模型地址;
  5. CrewTask 未绑定 Agent Task 实例必须指定agent参数,无执行主体会直接运行异常,每个任务必须明确归属智能体。

九、全文总结

本次四组实战练习完整覆盖两大多智能体框架入门核心能力:

  1. LangGraph 方向:掌握条件分支路由、多级层级路由、全局 State 状态流转,理解状态图编排底层逻辑,适配工单流转类业务;
  2. CrewAI 方向:掌握Agent 人设定义、Task 任务创建、任务与智能体绑定,学会轻量化多智能体团队基础搭建;

通过同一业务场景横向对比两个框架优劣、边界与选型思路,不止完成课后作业,同时建立多智能体架构选型思维,后续可进阶拓展混合架构、工具调用、持久化部署、人机协同等深度内容。

更多推荐