crew AI框架的理念是模拟团队协作的方式来实现特定目标。

agent是人，task是事，crew/flow是组织结构。人和组织有使用工具的能力，所以agent和crew可以增加tools。人和组织有经验，所以有memory和knowledge特性。做事情的方式多种多样，有多角色协作的crew模式，也有严格按照特定工作流的flow模式，不论是谁来做，都是以一个个具体的任务为单位进行协作的，所以单个任务的核心是task的设计。

基础内容可参考crew AI笔记[4] - 第一个crew AI代码

那么参考crew AI框架，能够完成基本agent设计以后，如何进行优化呢？这需要对各个特性进行更深入的理解，并结合实际业务场景灵活运用。

 

agent（能干嘛）

• role：是角色定位，有啥功能？有啥特长？
• goal：是你作为这个角色应该实现的
• backstory：为这个员工提供背景描述，增添个性
• tools：拥有的额外能力【可选，个人强烈推荐】
• knowledge_sources：拥有的知识【可选，但很常用】

上述是基础特性，除此之外呢？

allow_delegation详解（领导力）

如果一个agent使能了allow_delegation特性，代表他在工作中能够主动求助：这不是我的专场，我需要问下团队里其他专家。

我个人这么理解：A是个领导者，委托别人干活，自己汇报成果。

具体而言他可能会有如下动作：

1. 主动求助：当A发现执行过程中缺乏必要的工具、信息、专业知识时，他会主动识别这个不足。
2. 委托任务：当A发现自己不能时候，会把部分或者全部任务委托给团队中另一个智能体（假设为B）
3. 接受反馈：B（员工）完成任务后，将结果传给A
4. 整合完成：A利用（窃取）B的成果，继续完成其最初被分配的任务

啥时候要启用这个特性呢？

1. 多角色协作：你如果定义某个agent为项目经理，必须开启
2. 多领域专业技能：如果一个任务涉及多个专业领域，单个agent无法覆盖
3. 高复杂度任务：任务复杂，单个智能体可能出错或者卡住。开启后增加自我纠正和求助重试的机会。提高容错性和系统优化的可能。
4. 高自动化任务：希望系统自动分配，减少人工干预。比如客服系统【副作用是太过自主，没法有效控制信号流】

特别的：低复杂度或者顺序任务，需要严格控制执行流时，必须关闭这个特性

allow_code_execution（自主改造）

这是一个关于 CrewAI 最强大也是最需要谨慎对待的特性之一，正确使用它能使自己的agent从“信息处理者”升级为“问题终结者”

有啥用？

1. 自动生产脚本：赋予了agent自主coding、自主执行、自主解读python代码的能力【像病毒一样】
2. 自我进化：根据脚本执行情况调整后续的推理和行动，实现一个“行动、观察、学习、改善”的循环

风险提示：

• 安全性是第一考虑因素：强烈建议使用沙盒环境执行代码
• 必须安装docker环境
• LLM的coding能力要强：生产弱智代码，反而是个灾难。

    allow_code_execution=True,   # 核心设置：允许代码执行
    verbose=True 

啥时候启用这个特征？

任务涉及计算、数据验证、精确逻辑控制时需要

给个测试代码

import os
from crewai import Agent, Task, Crew, LLM
​
LLM_QW = LLM(
    model='dashscope/qwen-plus', 
    base_url='https://dashscope.aliyuncs.com/compatible-mode/v1', 
    api_key=os.environ.get("QWEN_API_KEY"), 
    stream=False, # 设置为同步模式
    temperature=0.7, # 控制模型输出的随机性
    request_timeout=240, # 增加请求超时时间到240秒
    max_retries=2,
)
​
​
# 创建可以执行代码的 Agent
code_exec_agent = Agent(
    role="Data Analyst",
    goal="Analyze data and generate insights using code execution",
    backstory="Experienced data analyst with a strong background in Python and data analysis",
    allow_code_execution=True, # 开启以后自动写脚本自动测试
    code_execution_mode="unsafe", # 默认safe模式，但需要配置docker环境
    llm=LLM_QW,
    verbose=True
)
​
# 创建任务
data_analysis_task = Task(
    description="Analyze the sales data and calculate the growth rate for the last quarter. The data is provided as a list of monthly sales figures: [12000, 15000, 18000, 20000].",
    expected_output="A string containing the growth rate calculation and the final growth rate percentage.",
    agent=code_exec_agent
)
​
# 创建 Crew
crew = Crew(
    agents=[code_exec_agent],
    tasks=[data_analysis_task],
    verbose=True
)
​
# 执行
try:
    result = crew.kickoff()
    print("Crew execution completed successfully!")
    print("Result:", result)
except Exception as e:
    print(f"Crew execution failed: {str(e)}")

task（要干嘛）

task是agent要完成的任务，task定义了执行所需要的细节。Task 是 整个 CrewAI 流程的核心单位，设计好坏直接影响最终效果。按照官方的说法，80%的设计应该围绕task进行。

基础属性

属性	说明	示例
description	任务详细描述	"分析销售数据，计算增长率"
expected_output	期望输出格式	"返回一个包含增长率的字符串"
agent	执行该任务的 Agent	data_analyst_agent
context	上下文依赖	[research_task]
output_file	输出到文件	"report.md"
output_json	输出为 JSON	{"growth_rate": 0.25}
output_pydantic	输出为 Pydantic 模型	自定义数据结构

task的输出

task_output = task.output

可以仔细观察output的细节，包括：描述、摘要、原始输出、pydantic输出、json输出等

context - 上下文依赖

启用上下文依赖关系，确保任务按顺序执行，该任务的启动依赖前序任务的结果。

context=[research_task]，等待research_task结束才会开启这个task

# 示例：撰写报告前必须先完成研究
write_report_task = Task(
    description="Write a report based on research findings",
    expected_output="A comprehensive report",
    agent=writer_agent,
    context=[research_task]  # 等待 research_task 完成
)

async_execution - 异步执行

对于可以并行的任务，可以开启这个特性，对性能提升极为有用

async_execution = True

适用场景：

• 数据采集
• 多源信息获取
• 不相互依赖的任务

callback - 任务回调

任务结束后自动触发回调函数

guardrail - 任务护栏（关键质量控制）

任务传递过程中提供了验证和转换的机制，有助于提高数据质量，并筛出不符合规范的数据

1、定义一个验证函数

2、在task中通过guardial赋值，使能验证函数

对验证函数的要求：

函数签名:

• 必须接受一个参数（任务输出）
• 应返回一个(bool, Any)的元组
• 推荐使用类型提示，但可选

返回值:

• 成功: 返回 (True, validated_result)
• 失败: 返回 (False, error_details)

示例：

from typing import Tuple, Union, Dict, Any
​
def validate_blog_content(result: str) -> Tuple[bool, Union[Dict[str, Any], str]]:
    """Validate blog content meets requirements."""
    try:
        # Check word count
        word_count = len(result.split())
        if word_count > 200:
            return (False, {
                "error": "Blog content exceeds 200 words",
                "code": "WORD_COUNT_ERROR",
                "context": {"word_count": word_count}
            })
​
        # Additional validation logic here
        return (True, result.strip())
    except Exception as e:
        return (False, {
            "error": "Unexpected error during validation",
            "code": "SYSTEM_ERROR"
        })
​
blog_task = Task(
    description="Write a blog post about AI",
    expected_output="A blog post under 200 words",
    agent=blog_agent,
    guardrail=validate_blog_content  # Add the guardrail function
)

护栏优势：

• 提高输出一致性
• 过滤无效/低质输出
• 实现自动化质量控制

crew（怎么干）

Crew 是多个 Agent 协作完成任务的组织形式。

基础属性

属性	说明	示例
agents	参与的 Agent 列表	[researcher, writer, reviewer]
tasks	要执行的任务列表	[research_task, write_task]
verbose	日志级别	True（详细日志）
process	执行流程类型	Process.sequential 或 Process.hierarchical

执行流程模式（活动的组织形式）

1. Process.sequential（顺序执行）

任务按顺序执行，前一个完成后一个才开始。

from crewai import Process

crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    process=Process.sequential  # 顺序执行
)

适用：逻辑依赖强的任务（如先研究后写作）

2. Process.hierarchical（分层执行）

自动分配任务给最合适的 Agent，适合复杂项目。

crew = Crew(
    agents=[senior_manager, junior_agent],
    tasks=[complex_task],
    process=Process.hierarchical,
    manager_llm=LLM_QW  # 指定管理用的 LLM
)

适用：多角色、高复杂度项目

3. Process.parallel（并行执行）

所有任务同时执行（需 async_execution=True）

crew = Crew(
    agents=[agent1, agent2],
    tasks=[task1, task2],
    process=Process.parallel
)

适用：独立任务，追求速度

Flow（工作流）—— 精确控制流程

适用场景：需要严格控制执行顺序的复杂流程。

详细内容参考： crew AI笔记[7] - flow特性示例

总结：CrewAI 设计哲学

组件	设计要点
Agent	角色清晰、能力明确、可求助、可编码
Task	描述精准、输出明确、有上下文、有护栏
Crew	流程合理、并行优化、自动调度
Flow	顺序可控、适合复杂流程

参考链接

win10的docker启动不了怎么办