基于 CrewAI 开发

CrewAI 是一个用于构建多 Agent 协作系统的 Python 框架，特别适合需要多个专业角色协同完成复杂任务的场景。云开发提供了 cloudbase-agent-crewai 适配器，让 CrewAI Agent 可以无缝对接 AG-UI 协议。

前置条件

• Python 3.9+
• 已开通云开发环境
• 已配置大模型（见 接入大模型）
• 已创建 API Key（获取地址）

安装依赖

pip install cloudbase-agent-crewai cloudbase-agent-server crewai crewai-tools

快速开始

基础示例

from cloudbase_agent.crewai import CrewAIAgent
from cloudbase_agent.server import create_agent_server
from crewai import Agent, Task, Crew

# 创建 Agent
researcher = Agent(
    role="研究员",
    goal="深入研究给定主题并提供详细分析",
    backstory="你是一位经验丰富的研究员，擅长收集和分析信息。",
    verbose=True,
)

writer = Agent(
    role="作家",
    goal="将研究结果转化为易于理解的文章",
    backstory="你是一位专业作家，擅长将复杂信息转化为通俗易懂的内容。",
    verbose=True,
)

# 创建任务
research_task = Task(
    description="研究 {topic} 的最新发展和趋势",
    expected_output="一份详细的研究报告",
    agent=researcher,
)

writing_task = Task(
    description="基于研究报告撰写一篇科普文章",
    expected_output="一篇 500 字左右的科普文章",
    agent=writer,
)

# 创建 Crew
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    verbose=True,
)

# 创建 CloudBase Agent
agent = CrewAIAgent(
    name="ResearchCrew",
    description="研究和写作团队",
    crew=crew,
)

# 创建服务
app = create_agent_server(agent)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=3000)

带工具的 Agent

from cloudbase_agent.crewai import CrewAIAgent
from cloudbase_agent.server import create_agent_server
from crewai import Agent, Task, Crew
from crewai_tools import SerperDevTool, WebsiteSearchTool

# 创建工具
search_tool = SerperDevTool()
website_tool = WebsiteSearchTool()

# 创建带工具的 Agent
researcher = Agent(
    role="网络研究员",
    goal="使用搜索工具收集最新信息",
    backstory="你是一位精通网络搜索的研究员。",
    tools=[search_tool, website_tool],
    verbose=True,
)

analyst = Agent(
    role="数据分析师",
    goal="分析收集到的信息并提取关键洞察",
    backstory="你是一位数据分析专家，擅长从大量信息中提取价值。",
    verbose=True,
)

# 创建任务
search_task = Task(
    description="搜索关于 {topic} 的最新新闻和文章",
    expected_output="收集到的相关信息列表",
    agent=researcher,
)

analysis_task = Task(
    description="分析搜索结果，提取关键信息和趋势",
    expected_output="分析报告",
    agent=analyst,
)

# 创建 Crew
crew = Crew(
    agents=[researcher, analyst],
    tasks=[search_task, analysis_task],
    verbose=True,
)

agent = CrewAIAgent(
    name="ResearchAnalysisCrew",
    description="研究分析团队",
    crew=crew,
)

app = create_agent_server(agent)

核心概念

Agent（代理）

Agent 是 CrewAI 中的核心角色，每个 Agent 都有特定的职责：

from crewai import Agent

agent = Agent(
    role="角色名称",           # 角色定义
    goal="目标描述",           # 要达成的目标
    backstory="背景故事",      # 角色背景，影响行为风格
    tools=[],                  # 可用工具列表
    llm=None,                  # 自定义 LLM（可选）
    verbose=True,              # 是否输出详细日志
    allow_delegation=True,     # 是否允许委托任务给其他 Agent
    max_iter=15,               # 最大迭代次数
)

Task（任务）

Task 定义了 Agent 需要完成的具体工作：

from crewai import Task

task = Task(
    description="任务描述，支持 {variable} 变量",
    expected_output="期望的输出格式",
    agent=agent,               # 负责执行的 Agent
    context=[other_task],      # 依赖的其他任务（可选）
    tools=[],                  # 任务专用工具（可选）
    async_execution=False,     # 是否异步执行
)

Crew（团队）

Crew 是 Agent 和 Task 的组合，定义了协作方式：

from crewai import Crew, Process

crew = Crew(
    agents=[agent1, agent2],
    tasks=[task1, task2],
    process=Process.sequential,  # 执行方式：sequential 或 hierarchical
    verbose=True,
    manager_llm=None,            # hierarchical 模式下的管理者 LLM
)

高级用法

层级执行模式

from crewai import Crew, Process
from langchain_openai import ChatOpenAI
import os

# 使用云开发内置端点创建管理者 LLM
manager_llm = ChatOpenAI(
    model="deepseek-v4-flash",
    api_key=os.environ.get("TCB_API_KEY"),
    base_url=f"https://{os.environ.get('TCB_ENV_ID')}.api.tcloudbasegateway.com/v1/ai/cloudbase",
)

# 创建层级 Crew
crew = Crew(
    agents=[researcher, writer, analyst],
    tasks=[research_task, writing_task, analysis_task],
    process=Process.hierarchical,
    manager_llm=manager_llm,
    verbose=True,
)

自定义 LLM

云开发内置了多种大模型，通过统一的 HTTP 端点访问：

from crewai import Agent
from langchain_openai import ChatOpenAI
import os

env_id = os.environ.get("TCB_ENV_ID")
api_key = os.environ.get("TCB_API_KEY")

# 使用腾讯混元模型
hunyuan_llm = ChatOpenAI(
    model="deepseek-v4-flash",
    api_key=api_key,
    base_url=f"https://{env_id}.api.tcloudbasegateway.com/v1/ai/cloudbase",
)

# 使用 DeepSeek 模型
deepseek_llm = ChatOpenAI(
    model="deepseek-r1-0528",  # 可选：deepseek-r1-0528, deepseek-v3-0324, deepseek-v3.2
    api_key=api_key,
    base_url=f"https://{env_id}.api.tcloudbasegateway.com/v1/ai/cloudbase",
)

agent = Agent(
    role="助手",
    goal="帮助用户完成任务",
    backstory="你是一个智能助手。",
    llm=hunyuan_llm,
)

如需使用外部模型 API，也可以自行配置：

from langchain_openai import ChatOpenAI

# OpenAI（需自行获取 API Key）
openai_llm = ChatOpenAI(
    model="gpt-4",
    api_key=os.environ.get("OPENAI_API_KEY"),
)

任务依赖

# 任务 2 依赖任务 1 的结果
task1 = Task(
    description="收集数据",
    expected_output="数据列表",
    agent=collector,
)

task2 = Task(
    description="分析数据",
    expected_output="分析报告",
    agent=analyst,
    context=[task1],  # 依赖 task1
)

回调处理

from cloudbase_agent.crewai import CrewAIAgent, CopilotKitState

def on_step_start(step_name: str, state: CopilotKitState):
    print(f"开始执行步骤: {step_name}")

def on_step_end(step_name: str, result: str, state: CopilotKitState):
    print(f"步骤完成: {step_name}, 结果: {result}")

agent = CrewAIAgent(
    name="CallbackCrew",
    description="带回调的团队",
    crew=crew,
    on_step_start=on_step_start,
    on_step_end=on_step_end,
)

流式输出

CrewAI Agent 默认支持流式输出，通过 copilotkit_stream 实现：

from cloudbase_agent.crewai import CrewAIAgent, copilotkit_stream

agent = CrewAIAgent(
    name="StreamCrew",
    description="流式输出团队",
    crew=crew,
)

# 流式输出会自动处理，客户端会收到实时的执行过程

常用工具

CrewAI 提供了丰富的内置工具：

from crewai_tools import (
    SerperDevTool,      # Google 搜索
    WebsiteSearchTool,  # 网站内容搜索
    FileReadTool,       # 文件读取
    DirectoryReadTool,  # 目录读取
    CodeInterpreterTool,# 代码执行
    PDFSearchTool,      # PDF 搜索
    YoutubeVideoSearchTool,  # YouTube 搜索
)

自定义工具

from crewai_tools import BaseTool
from pydantic import BaseModel, Field

class WeatherInput(BaseModel):
    city: str = Field(description="城市名称")

class WeatherTool(BaseTool):
    name: str = "get_weather"
    description: str = "获取指定城市的天气信息"
    args_schema: type[BaseModel] = WeatherInput

    def _run(self, city: str) -> str:
        # 实际项目中调用天气 API
        return f"{city}：25°C，晴天"

weather_tool = WeatherTool()

最佳实践

1. 角色设计

• 每个 Agent 应该有明确的职责边界
• 使用详细的 backstory 来引导 Agent 的行为风格
• 避免创建职责重叠的 Agent

2. 任务分解

• 将复杂任务分解为多个小任务
• 合理设置任务依赖关系
• 使用 context 传递上下文信息

3. 资源管理

# 设置最大迭代次数避免无限循环
agent = Agent(
    role="助手",
    goal="完成任务",
    backstory="你是一个助手。",
    max_iter=10,  # 最多迭代 10 次
)

4. 错误处理

from cloudbase_agent.crewai import CrewAIAgent

def on_error(error: Exception, state):
    print(f"执行出错: {error}")
    return {"error": str(error)}

agent = CrewAIAgent(
    name="ErrorHandlingCrew",
    description="带错误处理的团队",
    crew=crew,
    on_error=on_error,
)

部署

云函数部署

由于 CrewAI 是 Python 框架，推荐使用云托管部署。如需使用云函数，需要使用 Python 运行时。

云托管部署

参考 云托管部署

Dockerfile 示例：

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt ./
RUN pip install -r requirements.txt --no-cache-dir

COPY . .

ENV PORT=3000
EXPOSE 3000

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "3000"]

requirements.txt：

cloudbase-agent-crewai
cloudbase-agent-server
crewai
crewai-tools
uvicorn

相关文档

• LangChain 开发指南
• LangGraph 开发指南
• AG-UI 协议