★ FEATURED ARTICLE 1. 为什么需要条件任务默认流程的局限1.1 CrewAI 的常规启动逻辑CrewAI 是我目前用的比较顺手的多智能体编排框架。它最核心的思路就是“一个 Crew 里放多个 Agent每个 Agent 负责一个 Task然后按顺序跑”。如果你用的是默认的Process.sequential启动后就是第一个 Agent 完成任务把结果交给第二个 Agent第二个再继续往下走直到所有任务执行完。这套机制在处理线性流程时很省心先查数据、再清洗、再分析、再生成报告一路从左到右。但现实里的业务很少是这么直来直去的。比如内容审核流程系统先把文章跑一遍风险判断如果判断为“低风险”接下来应该自动发布如果判断为“高风险”接下来要走人工复核。这两种后续动作是互斥的不可能同时执行。用默认的 CrewAI 启动方式要么你把两个任务都挂上结果每次都会既“发布”又“复核”这显然不符合业务要么你就拆成两个 Crew手动判断之后再启动其中一个——这其实就是我们说的“条件任务”在启动层面的核心价值。1.2 条件任务解决了什么问题条件任务简单说就是“满足条件才执行某个 Task不满足就跳过”。它让流程从“一根筋”变成了“决策树”让每个 Agent 不用再把时间浪费在不该自己处理的输入上。我举一个实际例子。之前有个需求是做客服工单自动分流用户提交一条工单先由一个分类 Agent 判断这是“退款问题”“技术故障”还是“商务合作”。如果分到退款就走退款处理 Crew如果分到技术故障就走技术排查 Crew。如果不用条件任务我只能把三类处理逻辑全部塞进同一个 Crew让每个 Agent 都跑一遍然后靠最后一个汇总 Agent 去选合适的结果。这不仅浪费 token还经常出现分类 Agent 与处理 Agent 结论不一致的情况。把条件判断放到启动流程之前让分类结果作为“开关”只启动一个后续 Crew整体流程又快又稳。这种设计还能显著降低流程复杂度。你不再需要在一个 Agent 的 Prompt 里写“如果前面是 A 你就做什么如果是 B 你就做什么”这种大段条件分支而是让每个 Agent 只专注自己的一小块业务。Agent 的 Prompt 短了模型理解难度低了最终输出质量自然也会提升。1.3 适合条件任务的几类典型场景这里我列过几个比较常见的使用场景基本都是我在项目里踩过或者见别人踩过的风险控制类流程如内容审核、支付风控、用户输入安全检查。先做风险评估根据风险等级决定是“放行”还是“人工介入”。路由分发类流程如客服工单、邮件分类、Bug 报告分流。先判断类型再决定交给哪个专业 Agent 处理。多模型调度类流程比如根据输入文本的语言决定调用中文模型还是英文模型根据任务复杂度决定是否调用更大参数的模型。动态任务链比如数据分析里如果数据量少于 100 条就走快速统计如果数据量大就启动分布式计算集群。虽然这种不一定用 CrewAI 实现但思路完全一致。说到底条件任务就是给 Agent 流程加一个“if”而这个“if”放在哪里、怎么写是接下来要展开的重点。2. 在 CrewAI 启动中接入条件控制三种可行姿势2.1 方案 A外部编排用 Python 逻辑控制多个 Crew 的启动这是最稳、也是我最早在项目里落地的方式。思路非常简单不使用单个 Crew 完成所有逻辑而是把流程拆成多个 Crew用普通的 Pythonif来决定接下来启动哪个 Crew。打个比方你不让员工自己决定“要不要下班”而是先派一个人去检查“今天工作是否完成”检查结果报告给经理经理再决定让不让这个人下班。这里的“经理”就是我们的 Python 主程序是外部控制者。这样做的好处很明显逻辑完全透明任何一步都能打断、调试、打印。不依赖 CrewAI 的版本特性不管框架怎么升级这段代码都能跑。每个 Crew 的职责最小化Agent 的 Prompt 更好写Context 更干净。缺点也有每次切换 Crew 都要重新初始化 Agent会有一定的启动开销而且前一个 Crew 的输出不会自动传进下一个 Crew你得手动把需要的内容通过inputs传进去。下面是一段最小化的伪代码结构from crewai import Agent, Task, Crew, Process # 第一个 Crew做分析与判断 analysis_agent Agent( role内容分析师, goal判断输入内容的风险等级, backstory你是一名严谨的内容审核员, verboseTrue ) analyze_task Task( description分析内容{content}输出JSON格式包含risk字段值为low或high, expected_output一个合法的JSON字符串, agentanalysis_agent ) analysis_crew Crew( agents[analysis_agent], tasks[analyze_task], processProcess.sequential ) first_result analysis_crew.kickoff(inputs{content: 需要审核的文章内容}) # 用普通 Python 逻辑判断 risk parse_risk_from_output(str(first_result)) # 自定义解析函数 if risk low: # 启动第二个 Crew自动发布 publish_crew Crew(...) publish_crew.kickoff(inputs{content: 原文}) else: # 启动第三个 Crew人工复核 review_crew Crew(...) review_crew.kickoff(inputs{content: 原文})这样写哪怕你对 CrewAI 的内部机制不熟也能一眼看懂流程走向。实际项目里我通常把每个 Crew 封装成函数然后在一个“流程编排”函数里做条件路由。代码很直白后期维护也方便。2.2 方案 B单 Crew 内使用条件 Task框架级能力CrewAI 其实是支持在单个 Crew 的任务链中设置条件的。不同版本里 API 略有差异有的叫condition参数有的提供专门的ConditionalTask类。因为框架迭代非常快建议你动手前先查一下当前版本的官方文档或源码里 Task 类的定义。条件 Task 的核心写法通常长这样from crewai import Task publish_task Task( description将内容正式发布, expected_output发布成功的确认信息, agentpublisher_agent, conditionlambda ctx: ctx.get(risk) low ) review_task Task( description对高风险内容进行人工复核, expected_output复核结论, agentreviewer_agent, conditionlambda ctx: ctx.get(risk) high )这里的ctx一般是前序任务解析后的输出具体格式取决于你怎么设置任务的output_pydantic或output_json。如果当前任务返回的不是结构化数据ctx就是纯字符串那你就得在 condition 里自己解析字符串比如用risk: low in str(ctx)这种糙办法。但我不推荐这么做容易踩坑后面会讲。条件任务的好处是在一个 Crew 内部保持上下文连续性前面 Agent 的信息可以直接被后面 Agent 引用不用人工传参。而且框架会自动判断哪些任务需要执行、哪些跳过日志里也会显示“Task skipped”整体流程更紧凑。不过要注意如果你用的 CrewAI 版本不支持condition参数或者你自定义了比较复杂的 Process这套写法可能不生效。所以我建议你先看官方文档确认。如果没有条件 Task 的话就退回方案 A。2.3 方案 C自定义 Process 实现复杂路由如果你觉得方案 A 太“割裂”、方案 B 的 condition 不够灵活CrewAI 也允许你继承Process类重写execute方法在任务执行器中自由插入if / else逻辑。这个方案适合有复杂依赖、需要精细控制每一个步骤的场景。大致思路是from crewai import Process class ConditionalProcess(Process): def execute(self, crew_outputNone): for task in self.tasks: if task.should_execute(crew_output): result task.execute(...) crew_output result return crew_output这么做的问题是你需要对 CrewAI 内部Task.execute的签名和输入输出有足够了解而且框架升级后很可能破坏你的自定义代码。除非你是在写工具库或做深度二次开发我一般不建议普通业务项目走这条路。同样的需求用方案 A 三个 Crew 就能解决没必要把自己绑死到框架内部。3. 实操案例搭建一个带条件任务的内容风险审核 Crew3.1 业务目标与参数设计为了把上面的思路落到能跑的代码上我带大家做一个完整案例。假设我们有这样一个任务输入一段用户提交的文章内容系统先自动判断是否存在敏感风险然后根据风险等级决定走“自动发布”还是“人工复核”流程。流程拆解如下用第一个 Crew 做分析其中第一个 Agent 负责提取文章关键信息第二个 Agent 负责根据关键信息判断风险输出 JSON。解析 JSON拿到risk字段。如果risk low启动发布 Crew。如果risk high启动人工复核 Crew。为什么要把“提取关键信息”和“风险判断”分成两个 Agent因为“风险判断”需要更严谨的输出如果让同一个 Agent 既做提取又做判断它容易在长篇大论中丢失重点。分给两个 Agent可以让第一个 Agent 只输出结构化要点第二个 Agent 基于要点做判断准确性会高很多。3.2 环境准备与基础配置首先安装 CrewAIpip install crewai如果你计划使用 OpenAI 的模型需要配置环境变量。Linux / macOS 下直接export OPENAI_API_KEY你的keyWindows 下set OPENAI_API_KEY你的key为了让结构化输出更稳定我们在创建 Agent 时可以设置temperature0.2max_tokens给足一些避免输出到一半被截断。完整代码文件我命名为content_review_crew.py。下面逐步贴代码并解释关键点。3.3 完整代码实现import json import re from crewai import Agent, Task, Crew, Process # ---------- 工具函数从 LLM 输出中提取 JSON ---------- def extract_json(text: str): 从可能夹杂 Markdown 的文本中提取 JSON 对象 match re.search(r\{.*\}, text, re.DOTALL) if not match: raise ValueError(f未找到 JSON{text}) return json.loads(match.group())这里extract_json很重要。很多情况下CrewAI 的 Agent 输出会带着 “json” 或解释性文字如果你直接json.loads(str(result))大概率会报错。用正则把第一个{...}提取出来再解析能省掉很多麻烦。接下来定义三个 Agent 和对应的任务。# ---------- 第一个 Crew分析与判断 ---------- extractor Agent( role关键信息提取员, goal从用户输入的内容中提取出所有关键事实、人物、时间、地点和涉及的行为描述, backstory你是一个专业的信息提取引擎只输出结构化要点不添加任何主观判断。, verboseTrue, temperature0.2, max_iter2 ) risk_analyst Agent( role风险审核员, goal根据关键信息判断内容是否存在风险等级输出 low 或 high, backstory你是内容安全专家判断标准严格宁可错杀也不放过风险。, verboseTrue, temperature0.2, max_iter2 ) extract_task Task( description从以下内容中提取关键信息\n{content}\n输出格式为JSON字段包括summary, keywords, entities, expected_output包含 summary、keywords、entities 三个字段的JSON, agentextractor ) analyze_task Task( description根据以下关键信息判断内容风险等级\n{key_info}\n输出JSON格式字段包括risk值为low或high、reason判断理由, expected_output包含 risk 和 reason 字段的JSON, agentrisk_analyst ) analysis_crew Crew( agents[extractor, risk_analyst], tasks[extract_task, analyze_task], processProcess.sequential, verboseTrue )这里的analyze_task的描述接收{key_info}也就是上一个extract_task的输出。CrewAI 在顺序执行时会把上一个任务的输出作为下一个任务description中的变量填充进去。更稳妥的做法是配置context[extract_task]这在不同版本里支持也不一样建议先按官方文档。如果不支持上下文传递就把key_info成型为从extract_task手动取。现在执行第一个 Crewcontent 某网友发布了一篇帖子声称某地发生了一起事件并附带了多张图片。 first_result analysis_crew.kickoff(inputs{content: content}) print(分析结果, first_result)注意kickoff返回的是一个任务输出对象不同版本可能有str()或者.output这样的属性。为了通用我们把它转成字符串result_text str(first_result) parsed extract_json(result_text) risk parsed.get(risk) print(判定结果, risk)下面就是条件分支启动# ---------- 第二个 Crew自动发布 ---------- publisher Agent( role发布专员, goal将内容正常发布到平台, backstory你是平台运营人员负责执行发布操作。, verboseTrue ) publish_task Task( description将以下内容发布到平台\n{content}\n请返回发布成功的结果, expected_output包含发布状态的信息, agentpublisher ) publish_crew Crew( agents[publisher], tasks[publish_task], processProcess.sequential ) # ---------- 第三个 Crew人工复核 ---------- reviewer Agent( role人工复核员, goal对高风险内容进行逐条复核并给出最终处理意见, backstory你是内容安全团队的高级审核员经验丰富。, verboseTrue ) review_task Task( description对以下高风险内容进行复核并给出最终处理意见\n{content}, expected_output包含复核结论的信息, agentreviewer ) review_crew Crew( agents[reviewer], tasks[review_task], processProcess.sequential ) if risk low: result publish_crew.kickoff(inputs{content: content}) print(自动发布结果, result) elif risk high: result review_crew.kickoff(inputs{content: content}) print(人工复核结果, result) else: print(风险等级无法识别需要人工介入。)这段代码已经可以在你本地跑通了。你只需要替换成自己的 API key 和内容即可。3.4 参数选择的几个细节max_iter2限制 Agent 迭代次数避免它因为反复思考而消耗太多 token。这个值可以根据任务难度调大但默认不要给太高否则出错时会很啰嗦。temperature0.2任务要求输出结构化 JSON温度太高会导致字段名飘忽不定我们要尽量让它稳定。如果是创意写作类任务可以调到 0.7 以上。expected_output务必写清楚“合法JSON字符串”或“仅输出JSON”这能显著提升命中率。每个 Agent 的backstory对输出风格有影响。例如“判断标准严格”会让它更容易给出high你要根据业务容忍度调整措辞。4. 踩坑记录与常见问题排查4.1 条件判断总是走错分支有段时间我拿到的risk一直是字符串low\n或者low 因为 LLM 输出里可能带换行或空格。后来我统一在解析时做清理risk str(parsed.get(risk, )).strip().lower() if risk low: ...另外有些模型不按你的 JSON schema 返回会变成{risk: Low}首字母大写。为了兼容一定要.lower()。更稳的做法是用enum或Literal配置在提示词中比如“risk 字段只能取值为小写字符串 low 或 high”。4.2 LLM 输出带 Markdown 导致 JSON 解析失败这是新手最容易遇到的问题。CrewAI 的 Agent 虽然你让它返回 JSON但它可能好心地加上“json”代码块或者末尾写一句“以上就是我的分析”。如果直接用json.loads必然报错。解决方式就是我上面的extract_json函数用正则提取第一个{...}。还有一个更严格的做法设置 Task 的output_json参数让 CrewAI 强制解析为 Pydantic 对象。不同版本字段不同需要查文档。如果没有这个参数就用正则兜底。如果输出里不是 JSON 而是纯文本那就要重新审视 Prompt 是否清晰。我习惯在description末尾加一句话“不要输出任何解释只输出JSON对象。”4.3 条件 Task 执行后后续任务拿不到上下文如果你使用的是方案 B 的条件 Task发现跳过后的任务拿不到前序任务的输出大概率是因为你没有把条件 Task 的condition返回值与后续任务的context关联起来。CrewAI 的上下文传递机制在不同版本之间有差异有些需要你显式声明context。经验是除非你非常熟悉当前版本否则不要依赖“自动上下文”。手动用inputs传递是最不容易出错的previous_output str(previous_crew.kickoff(...)) next_crew.kickoff(inputs{previous_data: previous_output})不建议在小规模项目上为了“优雅”去赌框架行为。4.4 多次启动 Crew 太慢token 消耗大方案 A 中每个 Crew 都是独立实例内部会重新初始化 Agent 的上下文如果条件分支非常多启动开销会呈线性增长。比如要做 10 次条件判断就启动 10 次 Crew每次都要调用模型接口体验确实不好。如果分支很多建议改用“路由 Agent”模式在同一个 Crew 里放一个路由 Agent它一次性把所有判断做掉然后输出一个包含多个分支结论的 JSON。后续再根据这个 JSON 决定是否再启动别的 Crew。相当于把多次判断压缩成一次。还有一种优化用“跳过任务”代替“不启动任务”即在一个 Crew 中定义所有任务但通过条件 Task 让无关任务直接跳过这样至少上下文是连续的少了几次 Crew 初始化的开销。当然这同样要求你的框架版本支持条件 Task。4.5 不要用条件过度的 Agent 判断条件判断本身也可能依赖 LLM比如“判断这个文本是否包含风险”。但 LLM 是有可能判断失误的。尤其是高风险场景你不能把最后一道审核也交给模型。所以你可以在实施方案时给自己留后路条件 Task 只负责“预筛”真正的高风险判断必须有人工兜底。我一般在条件分支里加一个“无法判断”的兜底路径if risk in (low, high): ... else: # 走人工处理流程 human_review_crew.kickoff(...)这个“无法判断”的路径往往能救你一命。4.6 注意多 Crew 之间无法共享会话如果你在第一个 Crew 里已经和某个 Agent 对话了很久突然切到第二个 Crew第二个 Crew 里的 Agent 不会知道你们聊了什么。这不是 bug是设计如此。你需要把关键信息通过inputs传过去。所以做方案 A 时最好定义一套“标准数据传递格式”比如统一用 JSON。第一个 Crew 的输出第二个 Crew 能直接解析第三个 Crew 也能解析这样流程才顺畅。5. 一些实操后的个人体会条件任务的正确用法是“把决策从 Agent 手里拿出来放到流程控制层”。你别指望让同一个 Agent 在一个 Prompt 里处理完所有分支逻辑那只会让输出越来越不可控。我现在的习惯是先列业务决策树每个节点尽量用普通 Python 判断实在需要 LLM 判断的地方才让 Agent 输出结构化结果。这样既保持了多智能体的灵活性又不会让流程变成一个不可解释的黑箱。另外日志一定要打全。每次启动 Crew 之前打印一下“当前判定结果即将启动哪个 Crew”。线上出问题的时候你能很快看到是哪个判断条件出了问题而不是对着模型输出猜半天。如果有条件把每个 Crew 的输入输出都存一份 JSON排查效率会高很多。这套方法我已经用在好几个实际项目里从工单分流到内容审核再到报告生成效果都很稳定。如果你在适配不同 CrewAI 版本时遇到怪问题大概率是 API 变了优先去翻对应版本的源码别盲猜。
阅读完成 · 觉得有帮助？
