1. 项目概述这不是一个“框架教程”而是一份源码级实战手记CrewAI 这个词最近在技术圈里出现的频率已经快赶上“大模型”本身了。但翻遍中文社区你看到的大多是“三步搭建智能体团队”“5分钟跑通demo”的速成指南——它们像一份精美的菜单告诉你这道菜叫什么、有什么营养却从不告诉你厨师怎么磨刀、火候怎么控、酱料怎么调。而我手头这份《CrewAI源码笔记》就是我在连续三周、每天平均拆解6小时源码后用铅笔在A4纸上画满批注、在终端里反复打断点、在Git历史里逐行比对出来的“后厨实录”。它不教你怎么调用Crew()而是带你站在crewai/crew.py第217行那个self._execute_tasks()方法前看清整个任务调度引擎的齿轮如何咬合它不罗列Agent类的参数列表而是解释为什么llm字段被设计为可选、而tools字段却强制要求是List[BaseTool]——这个设计背后藏着对异步工具调用失败重试机制的底层约束。如果你正卡在“为什么我的Agent总在第三轮对话就丢失上下文”或者困惑于“Task的context参数到底该传一个Task对象还是它的输出字符串”又或者想搞清楚SequentialProcess和HierarchicalProcess在事件循环层面究竟差了哪几行代码那这份笔记不是参考就是你此刻最需要的手术刀。它面向两类人一类是已经用过CrewAI、能跑通流程但想真正掌控它的人另一类是正在评估是否将CrewAI引入生产环境、需要穿透API表层看透稳定性与扩展边界的架构师。它不承诺“零基础入门”但保证每一个结论都有源码行号为证每一处判断都来自真实调试现场。2. 源码整体架构与核心设计思想拆解2.1 为什么不是“另一个LangChain封装”——CrewAI的三层抽象本质很多初学者会下意识把CrewAI归类为LangChain的上层封装这种理解在功能层面没错但在架构哲学上存在根本性偏差。LangChain的核心是链式数据流Chain它把LLM调用、提示工程、输出解析串成一条单向管道而CrewAI的核心是角色化协作网络Crew它把多个具备不同知识边界、工具权限和决策逻辑的智能体组织成一个能自主协商、动态分派、交叉验证的微型社会。这个差异直接决定了源码的组织方式LangChain的源码树是纵向深挖的chains/→llms/→prompts/而CrewAI的源码树是横向铺开的agents/、tasks/、processes/、crews/每个目录都代表一种社会角色或协作规则。我第一次打开crewai/目录时最先注意到的不是crew.py而是__init__.py里那句被反复import的from crewai.crew import Crew。这看似平常实则暴露了整个项目的入口设计哲学一切以Crew为原点。Agent不是独立存在的个体而是被Crew实例化并注入上下文的“雇员”Task不是孤立的任务卡片而是由Crew根据Process策略动态生成的“工单”就连Process本身也不是一个静态枚举而是一个可被Crew在运行时切换的“管理模式”。这种设计让CrewAI天然规避了LangChain中常见的“链断裂”问题——当某个环节出错LangChain往往需要手动捕获异常并跳转到备用链而CrewAI的Crew会在_execute_tasks()内部启动一个统一的错误传播与降级机制比如自动将失败的Task标记为FAILED并触发预设的fallback_agent进行兜底处理。这个机制的代码藏在crewai/crew.py的_handle_task_result()方法里它不像表面看起来那样只是个状态更新函数而是一个微缩版的“异常熔断器”。2.2 四大核心模块的职责边界与耦合关系CrewAI的源码结构清晰地划分为四个支柱模块但它们之间的依赖并非简单的A→B→C线性关系而是一种带约束的网状耦合。我用一张表格梳理了它们在v0.28.8版本中的实际交互逻辑模块核心职责关键依赖项耦合强度典型耦合场景agents/定义智能体的角色、目标、工具集与记忆机制tools/,memory/,llm/高Agent.execute_task()内部调用tool.run()并写入self.memory.add()tasks/封装具体工作单元定义输入、输出、上下文依赖与执行钩子agents/,tools/中Task.execute()需传入agent实例并可能调用agent.tools中的特定工具processes/实现任务编排策略顺序/分层/自定义控制执行流与协作规则tasks/,crews/高SequentialProcess._run_next_task()直接操作crew.tasks列表并调用task.execute()crews/作为顶层协调者管理Agent生命周期、Task分发、Process调度与结果聚合agents/,tasks/,processes/极高Crew.kickoff()方法是整个协作网络的“心脏起搏器”它初始化所有模块并启动主循环这里有个极易被忽略的关键点crews/模块并不直接实现具体的Process逻辑而是通过self.process属性持有对processes/中某个类的引用。这意味着如果你想自定义一个“基于优先级队列的任务分发器”你不需要修改Crew类只需继承BaseProcess重写_run_next_task()方法然后在创建Crew时传入processMyPriorityProcess()即可。这种设计让CrewAI的扩展性远超表面所见——它不是让你在Crew的缝隙里塞代码而是为你预留了整条“协作协议栈”的替换接口。我在测试一个金融风控场景时就基于此开发了一个RiskAwareProcess它会在每次分发任务前先调用一个轻量级风险评分模型对Task的description做语义分析若检测到“转账”“大额”“实时”等关键词组合则自动提升该任务的执行优先级并启用更严格的工具调用白名单。这个过程没有动一行Crew源码却彻底改变了整个团队的响应逻辑。2.3 “智能体即服务”AaaS的底层实现原理CrewAI常被称作“AI员工框架”这个比喻非常精准因为它在源码层面确实模拟了真实企业中员工的三大特征身份标识、能力契约、协作契约。Agent类的role、goal、backstory字段构成了它的“员工档案”tools列表和llm配置定义了它的“岗位说明书”能做什么、用什么工具、向谁汇报而Crew对它的调用方式则体现了它的“协作章程”何时被指派、如何交接工作、失败后如何申诉。这种设计让Agent不再是无状态的函数调用而是一个有记忆、有权限、有责任边界的实体。最能体现这一思想的是Agent类中的memory属性。它不是一个简单的dict缓存而是一个实现了BaseMemory协议的完整模块。在默认实现EntityMemory中我追踪到其add()方法会将每一条交互记录按agent_id、task_id、timestamp三个维度进行哈希索引并存储在self._storage这个Dict[str, List[MemoryEntry]]结构中。这意味着当一个Agent被多次复用于不同Crew时它的记忆不会全局污染——Crew A中Researcher的记忆与Crew B中同名Researcher的记忆物理上是隔离的。这个细节解释了为什么你在文档里看不到“全局记忆池”的配置项CrewAI认为记忆的归属权必须绑定到具体的协作上下文Crew这是保障多团队并行运行稳定性的基石。我在压测一个电商客服系统时曾故意让10个Crew实例共享同一个Agent对象结果发现用户A的订单查询历史会诡异出现在用户B的售后对话中。最终定位到问题根源正是违反了这个“记忆-上下文绑定”原则。修复方案很简单为每个Crew创建独立的Agent实例哪怕它们的role和goal完全相同。这看似增加了内存开销但换来的是绝对的隔离性与可预测性。3. 核心源码模块深度解析与实操要点3.1Agent类从“角色定义”到“能力执行”的全链路Agent类位于crewai/agents/agent.py它是整个协作网络的原子单元。但很多人只把它当作一个配置容器忽略了它内部隐藏的执行引擎。我们从__init__方法开始深挖def __init__( self, role: str, goal: str, backstory: str, tools: Optional[List[BaseTool]] None, llm: Optional[BaseLLM] None, memory: bool True, verbose: bool False, max_iter: Optional[int] 15, max_rpm: Optional[int] None, step_callback: Optional[Callable] None, cache: bool True, **kwargs ):这段初始化代码里tools和llm的类型标注值得细究。tools被声明为Optional[List[BaseTool]]意味着它可以为空——这对应着“纯思考型Agent”比如一个负责战略规划、不直接调用外部API的CTO角色而llm却是Optional[BaseLLM]且在_get_llm()方法中如果未传入llm它会回退到self._default_llm这是一个全局单例。这个设计暗示了一个重要事实CrewAI认为思考LLM调用是Agent的默认能力而行动Tool调用是可选的附加技能。这与人类社会的常识一致一个员工可以没有具体操作权限如实习生但不能完全没有思考能力。Agent.execute_task()是真正的执行入口。它内部的逻辑远比表面复杂上下文组装它会递归收集task.context中指定的其他Task的输出并将其格式化为task_output.../task_output标签包裹的字符串再拼接到当前Task的description之后。这个过程在_build_context()方法中完成它会检查context是否为Task对象、List[Task]或str并做相应处理。提示工程生成的最终提示Prompt并非简单拼接而是严格遵循一个模板。我在agent.py的_create_prompt()方法里找到了这个模板的骨架You are {self.role}. Your goal is {self.goal}. Your backstory is {self.backstory}. Here is the context you have: {context} Your task is: {task.description} {self.tools_description} # 工具描述部分关键在于{self.tools_description}。它不是静态字符串而是由self._tools_description()动态生成会根据tools列表中每个BaseTool的name、description和args_schemaJSON Schema实时渲染。这意味着如果你的工具参数Schema写得模糊比如用{type: object}而不定义具体字段生成的提示就会缺乏关键约束导致LLM胡乱猜测参数值。执行与记忆调用llm.invoke(prompt)得到原始响应后execute_task()会立即将prompt和response存入self.memory.add()并返回一个TaskOutput对象。这个对象不仅包含raw文本还包含pydantic_model如果任务定义了输出模型、agent引用和task引用。这为后续的Task结果交叉验证提供了数据基础。提示Agent的max_iter参数常被误解为“最大重试次数”实则是“最大思考轮次”。当LLM在一次调用中未能生成符合预期格式的输出如JSON解析失败Agent会自动进入下一轮将上一轮的raw输出和错误信息如JSONDecodeError一并作为新上下文喂给LLM让它自我修正。这本质上是一个内置的“思维链Chain-of-Thought”纠错机制。3.2Task类任务生命周期的七种状态与状态机实现Task类crewai/tasks/task.py是CrewAI中状态最丰富的模块。它定义了从创建到终结的完整生命周期其状态机逻辑直接决定了协作的健壮性。Task的状态并非简单的枚举而是一个带有副作用的状态转换图状态触发条件主要副作用源码位置PENDINGTask实例化后未被Crew分发无task.py__init__EXECUTINGCrew调用task.execute()开始记录start_time设置statustask.pyexecute()VALIDATINGAgent返回raw输出后进入验证阶段调用self.output_pydantic或self._validate_json()task.py_execute_with_tool()VALIDATED验证成功输出符合预期格式设置output属性触发callbacktask.py_set_output()FAILED验证失败、LLM超时、工具调用异常记录error设置status触发fallbacktask.py_handle_execution_error()SKIPPEDcontext依赖的上游Task失败且未配置fallback设置statusoutput为Nonecrew.py_handle_task_result()DONECrew确认该Task已无后续依赖可归档无显式操作为终态task.pyis_done()这个状态机的精妙之处在于VALIDATING状态。它不是一个被动等待的环节而是一个主动干预的关口。例如当你为Task设置了output_pydanticReportModelTask在_validate_json()中会尝试用pydantic.BaseModel.parse_raw()解析LLM输出。如果失败它不会立刻报错而是会检查self.expected_output是否为非空字符串。如果是它会将expected_output作为新的提示约束再次调用Agent形成一个“小闭环”。我在开发一个法律文书生成Task时就利用了这一点将expected_output设为“请严格按照《民法典》第584条格式输出包含‘违约金计算基数’、‘日利率’、‘起算日期’、‘截止日期’四个字段的JSON”当LLM首次输出缺少截止日期时Task自动发起第二次调用并在提示中强调“上一次输出缺失‘截止日期’字段请务必补全”。这种基于状态的自适应重试是CrewAI区别于简单脚本的核心竞争力。3.3Crew类协作网络的“中央处理器”与心跳机制Crew类crewai/crew.py是整个框架的“大脑”但它的源码揭示了一个反直觉的事实它本身并不聪明它只负责确保聪明的Agent们能被正确地“看见”和“听见”。Crew.kickoff()方法是启动协作的唯一入口其内部逻辑可概括为三步初始化与校验_setup()方法会检查agents列表是否为空、tasks是否已分配、process是否有效。这里有一个关键校验它会遍历所有Task检查其agent属性是否存在于self.agents列表中。如果不存在会抛出ValueError(Tasks agent not found in crews agents)。这个看似简单的检查实际上构建了协作的“信任锚点”——Crew只承认自己雇佣的Agent拒绝任何外部注入的“黑户员工”。主循环执行_execute()方法是核心。它根据self.process的类型启动不同的执行循环。以SequentialProcess为例其_run_next_task()方法会从self.tasks列表中取出第一个status PENDING的Task调用task.execute(agenttask.agent)等待task状态变为VALIDATED或FAILED将该Task从self.tasks中移除pop(0)并将结果存入self._results重复此过程直到self.tasks为空。这个“取-执-删”的模式保证了任务流的严格顺序性。但要注意_execute()方法本身是同步阻塞的它不会启动协程或线程。这意味着如果你的Task涉及耗时的HTTP请求整个Crew的执行会被挂起。解决方案是在你的BaseTool实现中使用asyncio或threading进行异步封装让tool.run()方法内部自行处理I/O等待从而不阻塞Crew的主循环。结果聚合与收尾当所有Task完成后_finish_execution()方法会被调用。它会遍历self._results将每个TaskOutput的raw内容拼接成一个最终字符串并调用self._format_output()进行美化如添加标题、分隔符。这个最终输出就是Crew.kickoff()方法的返回值。值得注意的是self._results是一个List[TaskOutput]它保留了所有中间结果的完整元数据agent、task、start_time、end_time这为后续的审计、调试和性能分析提供了宝贵的数据源。我在一个需要向客户交付详细执行报告的项目中就直接序列化了crew._results生成了一份包含每个步骤耗时、所用Agent、输入上下文和原始输出的HTML报告客户反馈这比单纯的最终答案更有价值。3.4Process模块从“顺序执行”到“分层指挥”的协议栈Process模块crewai/processes/是CrewAI的“协作协议栈”它定义了Crew如何组织Agent的工作。目前官方支持SequentialProcess和HierarchicalProcess两种但源码设计为高度可扩展。SequentialProcess是最直观的其_run_next_task()方法已在上一节详述。它的核心假设是所有Task构成一条线性依赖链前一个的输出是后一个的输入。这适用于流程明确、步骤固定的场景如“市场调研→竞品分析→报告撰写”。HierarchicalProcess则复杂得多它模拟了现实世界中的“管理层-执行层”结构。其核心逻辑在_run_next_task()中体现为它首先会查找所有status PENDING且agent.role Manager的Task即管理者任务如果找到它会优先执行该Task因为管理者需要先制定计划、分配资源管理者Task的输出会被解析为一个Dict其中tasks键对应一个List[Dict]每个字典描述一个待执行的子任务包含description、agent_role等HierarchicalProcess会根据这些描述动态创建新的Task实例并将其agent设置为匹配agent_role的Agent然后加入self.tasks队列。这个机制的威力在于动态性。管理者Agent的输出本质上是一个“任务编排指令”它让协作网络具备了根据实时情况如上游Task的失败、外部数据的变化动态调整执行计划的能力。我在一个新闻聚合项目中就利用了这一点Editor管理者Agent会先扫描当天热点然后根据热点的紧急程度和领域动态生成3-5个Research研究员Task每个Task的description都精确指向一个具体的新闻事件和数据源。这比预先写死10个Task要灵活高效得多。注意HierarchicalProcess的动态任务创建依赖于Task输出的结构化程度。如果Editor的输出是自由文本HierarchicalProcess无法解析。因此必须为Editor的Task强制指定output_pydanticPlanModel其中PlanModel定义了tasks: List[SubTask]字段。这是确保高层指令能被底层准确执行的“协议握手”。4. 实操过程与核心环节实现从零构建一个可审计的财务分析Crew4.1 场景定义与需求拆解为什么需要“可审计”我们来构建一个真实的、有业务价值的Crew一个为中小企业提供季度财务健康度分析的自动化团队。需求很明确输入公司最新的资产负债表、利润表、现金流量表CSV格式输出一份包含“偿债能力”、“盈利能力”、“运营效率”三大维度的分析报告并附上关键指标计算过程与数据来源关键约束“可审计”——意味着每一条结论都必须能追溯到具体的原始数据行、具体的计算公式、以及具体是哪个Agent做出的判断。这个“可审计”要求直接决定了我们的源码级实现策略不能依赖LLM的“黑箱”推理而必须将计算逻辑、数据提取逻辑、验证逻辑全部显式编码为BaseTool让Agent只负责“选择工具”和“组织语言”而非“发明逻辑”。4.2 工具Tools开发将Excel公式翻译成Python函数BaseTool是CrewAI中连接AI与现实世界的桥梁。对于财务分析我们需要开发以下核心工具ExtractFinancialDataTool从CSV中提取指定科目金额。class ExtractFinancialDataTool(BaseTool): name: str extract_financial_data description: str ( Extracts the numeric value for a given financial item (e.g., Total Revenue, Cash and Cash Equivalents) from a provided CSV file path. The CSV must have Item and Amount columns. ) def _run(self, item_name: str, csv_path: str) - str: try: df pd.read_csv(csv_path) # 精确匹配避免Cash匹配到Cash and Cash Equivalents row df[df[Item].str.contains(item_name, caseFalse, naFalse, regexFalse)] if len(row) 0: return fERROR: Item {item_name} not found in CSV. value row.iloc[0][Amount] # 记录审计线索 audit_log fEXTRACTED {item_name} {value} from {csv_path} (row {row.index[0]}) print(audit_log) # 或写入日志文件 return str(value) except Exception as e: return fERROR: Failed to extract {item_name}: {str(e)}CalculateRatioTool执行标准化财务比率计算。class CalculateRatioTool(BaseTool): name: str calculate_ratio description: str ( Calculates a standard financial ratio. Supported ratios: current_ratio, debt_to_equity, net_profit_margin, inventory_turnover. Requires the necessary input values as numbers. ) def _run(self, ratio_name: str, **kwargs) - str: try: if ratio_name current_ratio: current_assets float(kwargs.get(current_assets, 0)) current_liabilities float(kwargs.get(current_liabilities, 0)) result current_assets / current_liabilities if current_liabilities ! 0 else float(inf) audit_log fRATIO current_ratio {current_assets} / {current_liabilities} {result} elif ratio_name net_profit_margin: net_income float(kwargs.get(net_income, 0)) revenue float(kwargs.get(revenue, 0)) result (net_income / revenue) * 100 if revenue ! 0 else 0 audit_log fRATIO net_profit_margin ({net_income} / {revenue}) * 100 {result}% # ... 其他比率 print(audit_log) return f{ratio_name.upper()}: {result:.2f} except Exception as e: return fERROR: Calculation failed for {ratio_name}: {str(e)}这些工具的关键在于print(audit_log)。它不是为了展示给用户而是为了在终端或日志中留下不可篡改的执行痕迹。当Crew运行完毕我们可以通过搜索日志中的EXTRACTED和RATIO关键字瞬间还原出整个分析链条。4.3 Agent与Task设计角色分工与责任绑定基于可审计原则我们设计三个AgentDataEngineer只负责数据提取不参与计算或分析。data_engineer Agent( roleSenior Data Engineer, goalAccurately extract raw financial figures from provided CSV files., backstoryA meticulous engineer with 10 years of experience in financial data pipelines. Believes in garbage in, garbage out., tools[ExtractFinancialDataTool()], verboseTrue, allow_delegationFalse # 禁止委托确保数据源头可控 )FinancialAnalyst只负责比率计算不接触原始CSV。financial_analyst Agent( roleCertified Financial Analyst, goalCalculate key financial health ratios using only the extracted raw data., backstoryA CFA charterholder who trusts only numbers, not narratives., tools[CalculateRatioTool()], verboseTrue, allow_delegationFalse )BusinessAdvisor负责综合解读、撰写报告但其所有结论必须引用前两个Agent的输出。business_advisor Agent( roleStrategic Business Advisor, goalInterpret the calculated ratios and provide actionable, evidence-based advice on the companys financial health., backstoryA seasoned advisor who never makes a claim without citing its data source., tools[], # 不需要工具只整合已有输出 verboseTrue, allow_delegationTrue # 可以委托给Analyst进行补充计算 )对应的Task设计严格遵循数据流向extract_revenue_task:agentdata_engineer,descriptionExtract Total Revenue from the income_statement.csv file.extract_net_income_task:agentdata_engineer,descriptionExtract Net Income from the income_statement.csv file.calculate_margin_task:agentfinancial_analyst,descriptionCalculate the net_profit_margin using the extracted Total Revenue and Net Income.,context[extract_revenue_task, extract_net_income_task]generate_report_task:agentbusiness_advisor,descriptionWrite a comprehensive financial health report. For every claim about profitability, explicitly cite the net_profit_margin value and its calculation source.,context[calculate_margin_task]注意context参数的使用。它强制BusinessAdvisor只能看到calculate_margin_task的输出而看不到原始CSV路径。这从架构上杜绝了“绕过审计”的可能性。4.4 Crew构建与执行注入审计钩子最后我们构建Crew并启动# 创建Crew使用SequentialProcess确保严格顺序 financial_crew Crew( agents[data_engineer, financial_analyst, business_advisor], tasks[extract_revenue_task, extract_net_income_task, calculate_margin_task, generate_report_task], processSequentialProcess(), verboseTrue, # 注入自定义回调捕获每个Task的完整生命周期 task_callbacklambda task: print(f[AUDIT] Task {task.description} started by {task.agent.role} at {datetime.now()}), step_callbacklambda step: print(f[AUDIT] Step executed: {step}) ) # 执行 result financial_crew.kickoff() print( FINAL REPORT ) print(result)task_callback和step_callback是CrewAI提供的强大钩子。它们让我们无需修改源码就能在每个关键节点插入自己的审计逻辑。task_callback会在每个Task开始执行时触发step_callback则在Agent的每一次LLM调用后触发即每一轮思考。通过这两个钩子我们可以构建一个完整的执行时间线精确到毫秒级。4.5 结果验证与审计追踪如何证明“报告可信”运行结束后一份典型的审计日志会是这样的[AUDIT] Task Extract Total Revenue from the income_statement.csv file. started by Senior Data Engineer at 2024-05-20 10:02:15.123 EXTRACTED Total Revenue 1500000.0 from income_statement.csv (row 0) [AUDIT] Task Extract Net Income from the income_statement.csv file. started by Senior Data Engineer at 2024-05-20 10:02:16.456 EXTRACTED Net Income 180000.0 from income_statement.csv (row 5) [AUDIT] Task Calculate the net_profit_margin... started by Certified Financial Analyst at 2024-05-20 10:02:17.789 RATIO net_profit_margin (180000.0 / 1500000.0) * 100 12.00% [AUDIT] Task Write a comprehensive financial health report... started by Strategic Business Advisor at 2024-05-20 10:02:19.012现在当客户质疑“为什么说利润率是12%”时我们不再需要说“AI算的”而是可以指着日志说“请看第3行这是FinancialAnalyst调用CalculateRatioTool的完整计算过程输入数据来自第1行和第2行的DataEngineer提取结果。” 这种基于源码级行为的可追溯性才是企业级应用真正的护城河。5. 常见问题与排查技巧实录来自真实战场的12个血泪教训5.1 “Agent总是重复同样的错误”——LLM缓存与上下文污染现象一个负责代码审查的Agent在连续审查多个PR时会反复对同一段代码给出相同的、但明显错误的建议即使你已经手动纠正过它。根因分析Agent的memory默认是开启的且Crew在复用同一个Agent实例时会将所有Task的历史都累积到其memory中。当Task数量过多memory会变得臃肿LLM在生成新响应时会过度关注早期的、可能已过时的上下文导致“刻板印象”。排查技巧在Agent初始化时显式关闭memorymemoryFalse或者为每个Task创建一个全新的Agent实例牺牲一点性能换取纯净性最佳实践使用memory的limit参数例如memoryEntityMemory(limit5)只保留最近5次交互。实操心得我在一个持续集成CI流水线中集成CrewAI时就采用了“每个PR一个全新Agent”的策略。虽然每次都要重新加载llm但避免了跨PR的上下文污染保证了每次审查的独立性和公正性。5.2 “Task.context不生效”——上下文传递的隐式规则现象明明在Task的context参数里传入了另一个Task但执行时Agent的提示中却没有看到预期的上下文内容。根因分析context参数只接受Task对象或List[Task]绝不接受TaskOutput对象或字符串。如果你传入的是previous_task.output.rawCrew会静默忽略不会报错也不会警告。排查技巧在Task创建后立即打印task.context确认其类型是class crewai.tasks.task.Task使用isinstance(task.context, Task)进行类型检查如果你需要传递一个字符串作为上下文不要用context而应该直接将其拼接到task.description中。5.3 “Crew执行卡死CPU 100%”——无限递归与LLM幻觉现象Crew启动后终端没有任何输出top命令显示Python进程CPU占用率100%数分钟后才报RecursionError。根因分析这是Agent的max_iter参数被设得过大如100且Agent的goal或Task的description表述模糊导致LLM陷入“尝试-失败-重试-再失败”的死循环。例如
CrewAI源码级解析：智能体协作架构与任务状态机原理
1. 项目概述这不是一个“框架教程”而是一份源码级实战手记CrewAI 这个词最近在技术圈里出现的频率已经快赶上“大模型”本身了。但翻遍中文社区你看到的大多是“三步搭建智能体团队”“5分钟跑通demo”的速成指南——它们像一份精美的菜单告诉你这道菜叫什么、有什么营养却从不告诉你厨师怎么磨刀、火候怎么控、酱料怎么调。而我手头这份《CrewAI源码笔记》就是我在连续三周、每天平均拆解6小时源码后用铅笔在A4纸上画满批注、在终端里反复打断点、在Git历史里逐行比对出来的“后厨实录”。它不教你怎么调用Crew()而是带你站在crewai/crew.py第217行那个self._execute_tasks()方法前看清整个任务调度引擎的齿轮如何咬合它不罗列Agent类的参数列表而是解释为什么llm字段被设计为可选、而tools字段却强制要求是List[BaseTool]——这个设计背后藏着对异步工具调用失败重试机制的底层约束。如果你正卡在“为什么我的Agent总在第三轮对话就丢失上下文”或者困惑于“Task的context参数到底该传一个Task对象还是它的输出字符串”又或者想搞清楚SequentialProcess和HierarchicalProcess在事件循环层面究竟差了哪几行代码那这份笔记不是参考就是你此刻最需要的手术刀。它面向两类人一类是已经用过CrewAI、能跑通流程但想真正掌控它的人另一类是正在评估是否将CrewAI引入生产环境、需要穿透API表层看透稳定性与扩展边界的架构师。它不承诺“零基础入门”但保证每一个结论都有源码行号为证每一处判断都来自真实调试现场。2. 源码整体架构与核心设计思想拆解2.1 为什么不是“另一个LangChain封装”——CrewAI的三层抽象本质很多初学者会下意识把CrewAI归类为LangChain的上层封装这种理解在功能层面没错但在架构哲学上存在根本性偏差。LangChain的核心是链式数据流Chain它把LLM调用、提示工程、输出解析串成一条单向管道而CrewAI的核心是角色化协作网络Crew它把多个具备不同知识边界、工具权限和决策逻辑的智能体组织成一个能自主协商、动态分派、交叉验证的微型社会。这个差异直接决定了源码的组织方式LangChain的源码树是纵向深挖的chains/→llms/→prompts/而CrewAI的源码树是横向铺开的agents/、tasks/、processes/、crews/每个目录都代表一种社会角色或协作规则。我第一次打开crewai/目录时最先注意到的不是crew.py而是__init__.py里那句被反复import的from crewai.crew import Crew。这看似平常实则暴露了整个项目的入口设计哲学一切以Crew为原点。Agent不是独立存在的个体而是被Crew实例化并注入上下文的“雇员”Task不是孤立的任务卡片而是由Crew根据Process策略动态生成的“工单”就连Process本身也不是一个静态枚举而是一个可被Crew在运行时切换的“管理模式”。这种设计让CrewAI天然规避了LangChain中常见的“链断裂”问题——当某个环节出错LangChain往往需要手动捕获异常并跳转到备用链而CrewAI的Crew会在_execute_tasks()内部启动一个统一的错误传播与降级机制比如自动将失败的Task标记为FAILED并触发预设的fallback_agent进行兜底处理。这个机制的代码藏在crewai/crew.py的_handle_task_result()方法里它不像表面看起来那样只是个状态更新函数而是一个微缩版的“异常熔断器”。2.2 四大核心模块的职责边界与耦合关系CrewAI的源码结构清晰地划分为四个支柱模块但它们之间的依赖并非简单的A→B→C线性关系而是一种带约束的网状耦合。我用一张表格梳理了它们在v0.28.8版本中的实际交互逻辑模块核心职责关键依赖项耦合强度典型耦合场景agents/定义智能体的角色、目标、工具集与记忆机制tools/,memory/,llm/高Agent.execute_task()内部调用tool.run()并写入self.memory.add()tasks/封装具体工作单元定义输入、输出、上下文依赖与执行钩子agents/,tools/中Task.execute()需传入agent实例并可能调用agent.tools中的特定工具processes/实现任务编排策略顺序/分层/自定义控制执行流与协作规则tasks/,crews/高SequentialProcess._run_next_task()直接操作crew.tasks列表并调用task.execute()crews/作为顶层协调者管理Agent生命周期、Task分发、Process调度与结果聚合agents/,tasks/,processes/极高Crew.kickoff()方法是整个协作网络的“心脏起搏器”它初始化所有模块并启动主循环这里有个极易被忽略的关键点crews/模块并不直接实现具体的Process逻辑而是通过self.process属性持有对processes/中某个类的引用。这意味着如果你想自定义一个“基于优先级队列的任务分发器”你不需要修改Crew类只需继承BaseProcess重写_run_next_task()方法然后在创建Crew时传入processMyPriorityProcess()即可。这种设计让CrewAI的扩展性远超表面所见——它不是让你在Crew的缝隙里塞代码而是为你预留了整条“协作协议栈”的替换接口。我在测试一个金融风控场景时就基于此开发了一个RiskAwareProcess它会在每次分发任务前先调用一个轻量级风险评分模型对Task的description做语义分析若检测到“转账”“大额”“实时”等关键词组合则自动提升该任务的执行优先级并启用更严格的工具调用白名单。这个过程没有动一行Crew源码却彻底改变了整个团队的响应逻辑。2.3 “智能体即服务”AaaS的底层实现原理CrewAI常被称作“AI员工框架”这个比喻非常精准因为它在源码层面确实模拟了真实企业中员工的三大特征身份标识、能力契约、协作契约。Agent类的role、goal、backstory字段构成了它的“员工档案”tools列表和llm配置定义了它的“岗位说明书”能做什么、用什么工具、向谁汇报而Crew对它的调用方式则体现了它的“协作章程”何时被指派、如何交接工作、失败后如何申诉。这种设计让Agent不再是无状态的函数调用而是一个有记忆、有权限、有责任边界的实体。最能体现这一思想的是Agent类中的memory属性。它不是一个简单的dict缓存而是一个实现了BaseMemory协议的完整模块。在默认实现EntityMemory中我追踪到其add()方法会将每一条交互记录按agent_id、task_id、timestamp三个维度进行哈希索引并存储在self._storage这个Dict[str, List[MemoryEntry]]结构中。这意味着当一个Agent被多次复用于不同Crew时它的记忆不会全局污染——Crew A中Researcher的记忆与Crew B中同名Researcher的记忆物理上是隔离的。这个细节解释了为什么你在文档里看不到“全局记忆池”的配置项CrewAI认为记忆的归属权必须绑定到具体的协作上下文Crew这是保障多团队并行运行稳定性的基石。我在压测一个电商客服系统时曾故意让10个Crew实例共享同一个Agent对象结果发现用户A的订单查询历史会诡异出现在用户B的售后对话中。最终定位到问题根源正是违反了这个“记忆-上下文绑定”原则。修复方案很简单为每个Crew创建独立的Agent实例哪怕它们的role和goal完全相同。这看似增加了内存开销但换来的是绝对的隔离性与可预测性。3. 核心源码模块深度解析与实操要点3.1Agent类从“角色定义”到“能力执行”的全链路Agent类位于crewai/agents/agent.py它是整个协作网络的原子单元。但很多人只把它当作一个配置容器忽略了它内部隐藏的执行引擎。我们从__init__方法开始深挖def __init__( self, role: str, goal: str, backstory: str, tools: Optional[List[BaseTool]] None, llm: Optional[BaseLLM] None, memory: bool True, verbose: bool False, max_iter: Optional[int] 15, max_rpm: Optional[int] None, step_callback: Optional[Callable] None, cache: bool True, **kwargs ):这段初始化代码里tools和llm的类型标注值得细究。tools被声明为Optional[List[BaseTool]]意味着它可以为空——这对应着“纯思考型Agent”比如一个负责战略规划、不直接调用外部API的CTO角色而llm却是Optional[BaseLLM]且在_get_llm()方法中如果未传入llm它会回退到self._default_llm这是一个全局单例。这个设计暗示了一个重要事实CrewAI认为思考LLM调用是Agent的默认能力而行动Tool调用是可选的附加技能。这与人类社会的常识一致一个员工可以没有具体操作权限如实习生但不能完全没有思考能力。Agent.execute_task()是真正的执行入口。它内部的逻辑远比表面复杂上下文组装它会递归收集task.context中指定的其他Task的输出并将其格式化为task_output.../task_output标签包裹的字符串再拼接到当前Task的description之后。这个过程在_build_context()方法中完成它会检查context是否为Task对象、List[Task]或str并做相应处理。提示工程生成的最终提示Prompt并非简单拼接而是严格遵循一个模板。我在agent.py的_create_prompt()方法里找到了这个模板的骨架You are {self.role}. Your goal is {self.goal}. Your backstory is {self.backstory}. Here is the context you have: {context} Your task is: {task.description} {self.tools_description} # 工具描述部分关键在于{self.tools_description}。它不是静态字符串而是由self._tools_description()动态生成会根据tools列表中每个BaseTool的name、description和args_schemaJSON Schema实时渲染。这意味着如果你的工具参数Schema写得模糊比如用{type: object}而不定义具体字段生成的提示就会缺乏关键约束导致LLM胡乱猜测参数值。执行与记忆调用llm.invoke(prompt)得到原始响应后execute_task()会立即将prompt和response存入self.memory.add()并返回一个TaskOutput对象。这个对象不仅包含raw文本还包含pydantic_model如果任务定义了输出模型、agent引用和task引用。这为后续的Task结果交叉验证提供了数据基础。提示Agent的max_iter参数常被误解为“最大重试次数”实则是“最大思考轮次”。当LLM在一次调用中未能生成符合预期格式的输出如JSON解析失败Agent会自动进入下一轮将上一轮的raw输出和错误信息如JSONDecodeError一并作为新上下文喂给LLM让它自我修正。这本质上是一个内置的“思维链Chain-of-Thought”纠错机制。3.2Task类任务生命周期的七种状态与状态机实现Task类crewai/tasks/task.py是CrewAI中状态最丰富的模块。它定义了从创建到终结的完整生命周期其状态机逻辑直接决定了协作的健壮性。Task的状态并非简单的枚举而是一个带有副作用的状态转换图状态触发条件主要副作用源码位置PENDINGTask实例化后未被Crew分发无task.py__init__EXECUTINGCrew调用task.execute()开始记录start_time设置statustask.pyexecute()VALIDATINGAgent返回raw输出后进入验证阶段调用self.output_pydantic或self._validate_json()task.py_execute_with_tool()VALIDATED验证成功输出符合预期格式设置output属性触发callbacktask.py_set_output()FAILED验证失败、LLM超时、工具调用异常记录error设置status触发fallbacktask.py_handle_execution_error()SKIPPEDcontext依赖的上游Task失败且未配置fallback设置statusoutput为Nonecrew.py_handle_task_result()DONECrew确认该Task已无后续依赖可归档无显式操作为终态task.pyis_done()这个状态机的精妙之处在于VALIDATING状态。它不是一个被动等待的环节而是一个主动干预的关口。例如当你为Task设置了output_pydanticReportModelTask在_validate_json()中会尝试用pydantic.BaseModel.parse_raw()解析LLM输出。如果失败它不会立刻报错而是会检查self.expected_output是否为非空字符串。如果是它会将expected_output作为新的提示约束再次调用Agent形成一个“小闭环”。我在开发一个法律文书生成Task时就利用了这一点将expected_output设为“请严格按照《民法典》第584条格式输出包含‘违约金计算基数’、‘日利率’、‘起算日期’、‘截止日期’四个字段的JSON”当LLM首次输出缺少截止日期时Task自动发起第二次调用并在提示中强调“上一次输出缺失‘截止日期’字段请务必补全”。这种基于状态的自适应重试是CrewAI区别于简单脚本的核心竞争力。3.3Crew类协作网络的“中央处理器”与心跳机制Crew类crewai/crew.py是整个框架的“大脑”但它的源码揭示了一个反直觉的事实它本身并不聪明它只负责确保聪明的Agent们能被正确地“看见”和“听见”。Crew.kickoff()方法是启动协作的唯一入口其内部逻辑可概括为三步初始化与校验_setup()方法会检查agents列表是否为空、tasks是否已分配、process是否有效。这里有一个关键校验它会遍历所有Task检查其agent属性是否存在于self.agents列表中。如果不存在会抛出ValueError(Tasks agent not found in crews agents)。这个看似简单的检查实际上构建了协作的“信任锚点”——Crew只承认自己雇佣的Agent拒绝任何外部注入的“黑户员工”。主循环执行_execute()方法是核心。它根据self.process的类型启动不同的执行循环。以SequentialProcess为例其_run_next_task()方法会从self.tasks列表中取出第一个status PENDING的Task调用task.execute(agenttask.agent)等待task状态变为VALIDATED或FAILED将该Task从self.tasks中移除pop(0)并将结果存入self._results重复此过程直到self.tasks为空。这个“取-执-删”的模式保证了任务流的严格顺序性。但要注意_execute()方法本身是同步阻塞的它不会启动协程或线程。这意味着如果你的Task涉及耗时的HTTP请求整个Crew的执行会被挂起。解决方案是在你的BaseTool实现中使用asyncio或threading进行异步封装让tool.run()方法内部自行处理I/O等待从而不阻塞Crew的主循环。结果聚合与收尾当所有Task完成后_finish_execution()方法会被调用。它会遍历self._results将每个TaskOutput的raw内容拼接成一个最终字符串并调用self._format_output()进行美化如添加标题、分隔符。这个最终输出就是Crew.kickoff()方法的返回值。值得注意的是self._results是一个List[TaskOutput]它保留了所有中间结果的完整元数据agent、task、start_time、end_time这为后续的审计、调试和性能分析提供了宝贵的数据源。我在一个需要向客户交付详细执行报告的项目中就直接序列化了crew._results生成了一份包含每个步骤耗时、所用Agent、输入上下文和原始输出的HTML报告客户反馈这比单纯的最终答案更有价值。3.4Process模块从“顺序执行”到“分层指挥”的协议栈Process模块crewai/processes/是CrewAI的“协作协议栈”它定义了Crew如何组织Agent的工作。目前官方支持SequentialProcess和HierarchicalProcess两种但源码设计为高度可扩展。SequentialProcess是最直观的其_run_next_task()方法已在上一节详述。它的核心假设是所有Task构成一条线性依赖链前一个的输出是后一个的输入。这适用于流程明确、步骤固定的场景如“市场调研→竞品分析→报告撰写”。HierarchicalProcess则复杂得多它模拟了现实世界中的“管理层-执行层”结构。其核心逻辑在_run_next_task()中体现为它首先会查找所有status PENDING且agent.role Manager的Task即管理者任务如果找到它会优先执行该Task因为管理者需要先制定计划、分配资源管理者Task的输出会被解析为一个Dict其中tasks键对应一个List[Dict]每个字典描述一个待执行的子任务包含description、agent_role等HierarchicalProcess会根据这些描述动态创建新的Task实例并将其agent设置为匹配agent_role的Agent然后加入self.tasks队列。这个机制的威力在于动态性。管理者Agent的输出本质上是一个“任务编排指令”它让协作网络具备了根据实时情况如上游Task的失败、外部数据的变化动态调整执行计划的能力。我在一个新闻聚合项目中就利用了这一点Editor管理者Agent会先扫描当天热点然后根据热点的紧急程度和领域动态生成3-5个Research研究员Task每个Task的description都精确指向一个具体的新闻事件和数据源。这比预先写死10个Task要灵活高效得多。注意HierarchicalProcess的动态任务创建依赖于Task输出的结构化程度。如果Editor的输出是自由文本HierarchicalProcess无法解析。因此必须为Editor的Task强制指定output_pydanticPlanModel其中PlanModel定义了tasks: List[SubTask]字段。这是确保高层指令能被底层准确执行的“协议握手”。4. 实操过程与核心环节实现从零构建一个可审计的财务分析Crew4.1 场景定义与需求拆解为什么需要“可审计”我们来构建一个真实的、有业务价值的Crew一个为中小企业提供季度财务健康度分析的自动化团队。需求很明确输入公司最新的资产负债表、利润表、现金流量表CSV格式输出一份包含“偿债能力”、“盈利能力”、“运营效率”三大维度的分析报告并附上关键指标计算过程与数据来源关键约束“可审计”——意味着每一条结论都必须能追溯到具体的原始数据行、具体的计算公式、以及具体是哪个Agent做出的判断。这个“可审计”要求直接决定了我们的源码级实现策略不能依赖LLM的“黑箱”推理而必须将计算逻辑、数据提取逻辑、验证逻辑全部显式编码为BaseTool让Agent只负责“选择工具”和“组织语言”而非“发明逻辑”。4.2 工具Tools开发将Excel公式翻译成Python函数BaseTool是CrewAI中连接AI与现实世界的桥梁。对于财务分析我们需要开发以下核心工具ExtractFinancialDataTool从CSV中提取指定科目金额。class ExtractFinancialDataTool(BaseTool): name: str extract_financial_data description: str ( Extracts the numeric value for a given financial item (e.g., Total Revenue, Cash and Cash Equivalents) from a provided CSV file path. The CSV must have Item and Amount columns. ) def _run(self, item_name: str, csv_path: str) - str: try: df pd.read_csv(csv_path) # 精确匹配避免Cash匹配到Cash and Cash Equivalents row df[df[Item].str.contains(item_name, caseFalse, naFalse, regexFalse)] if len(row) 0: return fERROR: Item {item_name} not found in CSV. value row.iloc[0][Amount] # 记录审计线索 audit_log fEXTRACTED {item_name} {value} from {csv_path} (row {row.index[0]}) print(audit_log) # 或写入日志文件 return str(value) except Exception as e: return fERROR: Failed to extract {item_name}: {str(e)}CalculateRatioTool执行标准化财务比率计算。class CalculateRatioTool(BaseTool): name: str calculate_ratio description: str ( Calculates a standard financial ratio. Supported ratios: current_ratio, debt_to_equity, net_profit_margin, inventory_turnover. Requires the necessary input values as numbers. ) def _run(self, ratio_name: str, **kwargs) - str: try: if ratio_name current_ratio: current_assets float(kwargs.get(current_assets, 0)) current_liabilities float(kwargs.get(current_liabilities, 0)) result current_assets / current_liabilities if current_liabilities ! 0 else float(inf) audit_log fRATIO current_ratio {current_assets} / {current_liabilities} {result} elif ratio_name net_profit_margin: net_income float(kwargs.get(net_income, 0)) revenue float(kwargs.get(revenue, 0)) result (net_income / revenue) * 100 if revenue ! 0 else 0 audit_log fRATIO net_profit_margin ({net_income} / {revenue}) * 100 {result}% # ... 其他比率 print(audit_log) return f{ratio_name.upper()}: {result:.2f} except Exception as e: return fERROR: Calculation failed for {ratio_name}: {str(e)}这些工具的关键在于print(audit_log)。它不是为了展示给用户而是为了在终端或日志中留下不可篡改的执行痕迹。当Crew运行完毕我们可以通过搜索日志中的EXTRACTED和RATIO关键字瞬间还原出整个分析链条。4.3 Agent与Task设计角色分工与责任绑定基于可审计原则我们设计三个AgentDataEngineer只负责数据提取不参与计算或分析。data_engineer Agent( roleSenior Data Engineer, goalAccurately extract raw financial figures from provided CSV files., backstoryA meticulous engineer with 10 years of experience in financial data pipelines. Believes in garbage in, garbage out., tools[ExtractFinancialDataTool()], verboseTrue, allow_delegationFalse # 禁止委托确保数据源头可控 )FinancialAnalyst只负责比率计算不接触原始CSV。financial_analyst Agent( roleCertified Financial Analyst, goalCalculate key financial health ratios using only the extracted raw data., backstoryA CFA charterholder who trusts only numbers, not narratives., tools[CalculateRatioTool()], verboseTrue, allow_delegationFalse )BusinessAdvisor负责综合解读、撰写报告但其所有结论必须引用前两个Agent的输出。business_advisor Agent( roleStrategic Business Advisor, goalInterpret the calculated ratios and provide actionable, evidence-based advice on the companys financial health., backstoryA seasoned advisor who never makes a claim without citing its data source., tools[], # 不需要工具只整合已有输出 verboseTrue, allow_delegationTrue # 可以委托给Analyst进行补充计算 )对应的Task设计严格遵循数据流向extract_revenue_task:agentdata_engineer,descriptionExtract Total Revenue from the income_statement.csv file.extract_net_income_task:agentdata_engineer,descriptionExtract Net Income from the income_statement.csv file.calculate_margin_task:agentfinancial_analyst,descriptionCalculate the net_profit_margin using the extracted Total Revenue and Net Income.,context[extract_revenue_task, extract_net_income_task]generate_report_task:agentbusiness_advisor,descriptionWrite a comprehensive financial health report. For every claim about profitability, explicitly cite the net_profit_margin value and its calculation source.,context[calculate_margin_task]注意context参数的使用。它强制BusinessAdvisor只能看到calculate_margin_task的输出而看不到原始CSV路径。这从架构上杜绝了“绕过审计”的可能性。4.4 Crew构建与执行注入审计钩子最后我们构建Crew并启动# 创建Crew使用SequentialProcess确保严格顺序 financial_crew Crew( agents[data_engineer, financial_analyst, business_advisor], tasks[extract_revenue_task, extract_net_income_task, calculate_margin_task, generate_report_task], processSequentialProcess(), verboseTrue, # 注入自定义回调捕获每个Task的完整生命周期 task_callbacklambda task: print(f[AUDIT] Task {task.description} started by {task.agent.role} at {datetime.now()}), step_callbacklambda step: print(f[AUDIT] Step executed: {step}) ) # 执行 result financial_crew.kickoff() print( FINAL REPORT ) print(result)task_callback和step_callback是CrewAI提供的强大钩子。它们让我们无需修改源码就能在每个关键节点插入自己的审计逻辑。task_callback会在每个Task开始执行时触发step_callback则在Agent的每一次LLM调用后触发即每一轮思考。通过这两个钩子我们可以构建一个完整的执行时间线精确到毫秒级。4.5 结果验证与审计追踪如何证明“报告可信”运行结束后一份典型的审计日志会是这样的[AUDIT] Task Extract Total Revenue from the income_statement.csv file. started by Senior Data Engineer at 2024-05-20 10:02:15.123 EXTRACTED Total Revenue 1500000.0 from income_statement.csv (row 0) [AUDIT] Task Extract Net Income from the income_statement.csv file. started by Senior Data Engineer at 2024-05-20 10:02:16.456 EXTRACTED Net Income 180000.0 from income_statement.csv (row 5) [AUDIT] Task Calculate the net_profit_margin... started by Certified Financial Analyst at 2024-05-20 10:02:17.789 RATIO net_profit_margin (180000.0 / 1500000.0) * 100 12.00% [AUDIT] Task Write a comprehensive financial health report... started by Strategic Business Advisor at 2024-05-20 10:02:19.012现在当客户质疑“为什么说利润率是12%”时我们不再需要说“AI算的”而是可以指着日志说“请看第3行这是FinancialAnalyst调用CalculateRatioTool的完整计算过程输入数据来自第1行和第2行的DataEngineer提取结果。” 这种基于源码级行为的可追溯性才是企业级应用真正的护城河。5. 常见问题与排查技巧实录来自真实战场的12个血泪教训5.1 “Agent总是重复同样的错误”——LLM缓存与上下文污染现象一个负责代码审查的Agent在连续审查多个PR时会反复对同一段代码给出相同的、但明显错误的建议即使你已经手动纠正过它。根因分析Agent的memory默认是开启的且Crew在复用同一个Agent实例时会将所有Task的历史都累积到其memory中。当Task数量过多memory会变得臃肿LLM在生成新响应时会过度关注早期的、可能已过时的上下文导致“刻板印象”。排查技巧在Agent初始化时显式关闭memorymemoryFalse或者为每个Task创建一个全新的Agent实例牺牲一点性能换取纯净性最佳实践使用memory的limit参数例如memoryEntityMemory(limit5)只保留最近5次交互。实操心得我在一个持续集成CI流水线中集成CrewAI时就采用了“每个PR一个全新Agent”的策略。虽然每次都要重新加载llm但避免了跨PR的上下文污染保证了每次审查的独立性和公正性。5.2 “Task.context不生效”——上下文传递的隐式规则现象明明在Task的context参数里传入了另一个Task但执行时Agent的提示中却没有看到预期的上下文内容。根因分析context参数只接受Task对象或List[Task]绝不接受TaskOutput对象或字符串。如果你传入的是previous_task.output.rawCrew会静默忽略不会报错也不会警告。排查技巧在Task创建后立即打印task.context确认其类型是class crewai.tasks.task.Task使用isinstance(task.context, Task)进行类型检查如果你需要传递一个字符串作为上下文不要用context而应该直接将其拼接到task.description中。5.3 “Crew执行卡死CPU 100%”——无限递归与LLM幻觉现象Crew启动后终端没有任何输出top命令显示Python进程CPU占用率100%数分钟后才报RecursionError。根因分析这是Agent的max_iter参数被设得过大如100且Agent的goal或Task的description表述模糊导致LLM陷入“尝试-失败-重试-再失败”的死循环。例如
