1. 项目背景
某科技公司的 HR 部门每周收到约 200 份简历,筛选初面的工作量大且主观性强。HR 经理想用 CrewAI 做一个"简历筛选助手":输入简历文本,输出候选人的技能匹配度、风险点和面试建议,然后用人系统自动归档。
第一版实现中,Agent 输出了这样一段文本:
“候选人张三,3年Python经验,做过2个数据分析项目。建议进入初面。沟通能力看起来不错。”
虽然内容看起来有道理,但下游系统完全无法解析——没有结构化的分数(匹配度到底是多少?),没有字段分隔(哪部分是技能评估?哪部分是风险?),更无法批量处理(200 份简历需要 200 次人工从文本中提取关键字段)。
核心痛点:大模型的输出本质是自然语言,天然"非结构化"。当 CrewAI 的输出需要被下游系统(前端、数据库、另一个 Agent)消费时,“看起来对"远远不够——必须"格式对、字段全、值合规”。
CrewAI 提供了三种结构化输出方式:Markdown(人类阅读友好)、JSON(程序解析友好)、Pydantic Model(强类型校验友好)。本章的目标是掌握expected_output与结构化输出的协同使用,让 Agent 的输出从"看起来像回事"变成"系统能直接消费"。
2. 项目设计
小胖(喝着奶茶):“大师,Agent 输出不就是一段文字吗?为什么还要搞结构化?我直接用正则提取不就行了?”
大师:“那你试试用正则提取’匹配度大概 70%-80% 吧’这句话。Agent 可能写成’70-80%‘,也可能写成’约七到八成’,甚至’匹配度较高’。你的正则需要覆盖所有这些变体——这本身就是个 NLP 问题。”
小白(翻开文档):“CrewAI 的 Task 有个expected_output参数。我之前以为它只是个’提示’,现在看文档说它实际上会被用于解析和校验?”
大师:“对。expected_output不止是给 Agent 看的目标描述,它也是 CrewAI 输出解析的’模板’。当你在expected_output里写了表格结构,Agent 会倾向于输出表格;当你在expected_output里定义了字段,Agent 会倾向于包含这些字段。如果再配合output_json或output_pydantic,CrewAI 会在模型输出后进行格式校验和解析。”
小胖:“那 Markdown、JSON、Pydantic 这三种格式怎么选?”
大师:“Markdown 适合人类阅读的最终产物——比如报告、邮件正文、攻略文档。JSON 适合给程序消费——比如 API 响应、数据库写入。Pydantic 适合给强类型系统消费——你需要保证字段类型、值范围和必填项的场景。”
小白:“Pydantic 的优势在哪?JSON 本身也可以有 schema 约束啊。”
大师:“Pydantic 的优势在于它不只是声明字段——它提供运行时校验。比如你声明score: int = Field(ge=0, le=100),Pydantic 会在解析时校验 score 是不是在 0-100 之间的整数。如果 Agent 输出'score': '八十五',Pydantic 会抛出 ValidationError——这个异常可以被 Guardrail 捕获并触发重试。JSON 做不到这种级别的校验。”
小胖:“那如果 Agent 输出的 JSON 格式不对怎么办?比如多了个逗号、少了引号?”
大师:“这就是结构化输出的核心挑战——模型的输出不是 100% 合规的。CrewAI 的策略是:用expected_output给模型一个’输出模板’降低格式错误概率,用output_pydantic做’最终校验’,解析失败时自动触发重试。但重试不是无限次的——max_retries参数控制。如果多次重试仍失败,最终的异常需要业务层兜底处理。”
小白:“那输出契约应该怎么设计?比如面向前端、后端、测试三个角色的输出。”
大师:“这是个好问题的进阶版本。向前端输出,字段名用 camelCase + 中文说明,附带显示建议(如’超过80分绿色显示’)。向后端输出,字段名用 snake_case + 英文,附带数据来源(‘该值来自 Agent 模型推理’)。向测试输出,附带置信度字段,让测试人员知道哪些字段需要重点人工核验。每个角色关心的维度不同,输出契约也不同。”
技术映射总结:Markdown 用于"给人看"(报告、摘要、文档),JSON 用于"给程序读"(API 响应、数据库写入),Pydantic Model 用于"给强类型系统用"(带校验、带默认值、带约束)。
expected_output是三者的"共同前奏"——它既指导 Agent 怎么写,又为后续解析提供模板。
3. 项目实战
3.1 实战目标
构建"简历筛选 Crew",输入简历文本,输出 Pydantic 结构化的候选人评估结果——包含技能匹配度、风险点和面试建议。对比 Markdown 输出和 Pydantic 输出的下游消费效率。
3.2 环境准备
resume-screener-crew/ ├── .env ├── main.py ├── models.py # Pydantic 输出模型定义 ├── agents.py ├── tasks.py ├── crew.py ├── sample_resumes/ # 测试用简历样本 └── outputs/依赖:crewai、pydantic、python-dotenv
3.3 分步实现
步骤1:定义 Pydantic 输出模型
目标:用 Pydantic 定义结构化的候选人评估输出格式。
# models.pyfrompydanticimportBaseModel,FieldfromtypingimportList,OptionalfromenumimportEnumclassRiskLevel(str,Enum):LOW="低"MEDIUM="中"HIGH="高"classInterviewDecision(str,Enum):RECOMMEND="推荐面试"CONSIDER="可考虑"NOT_RECOMMEND="不推荐"classSkillItem(BaseModel):"""单项技能评估"""name:str=Field(description="技能名称")required:bool=Field(description="是否为岗位必需技能")match_level:int=Field(ge=0,le=100,description="匹配度评分 0-100,100表示完全匹配")evidence:str=Field(description="匹配证据:简历中的具体描述或项目经验")classRiskItem(BaseModel):"""单项风险评估"""category:str=Field(description="风险类别,如'稳定性'、'技能缺口'、'薪资预期'")level:RiskLevel=Field(description="风险等级")detail:str=Field(description="风险描述")mitigation:Optional[str]=Field(default=None,description="缓解建议(如有)")classInterviewQuestion(BaseModel):"""面试建议问题"""question:str=Field(description="建议在面试中提出的问题")purpose:str=Field(description="提问目的,如'验证技术深度'、'评估沟通能力'")classCandidateEvaluation(BaseModel):"""简历筛选评估结果——Pydantic 输出模型"""candidate_name:str=Field(description="候选人姓名")overall_score:int=Field(ge=0,le=100,description="综合评分 0-100")decision:InterviewDecision=Field(description="面试决策")skill_assessment:List[SkillItem]=Field(description="技能评估列表,至少包含3项")highlights:List[str]=Field(description="候选人亮点,2-3条")risks:List[RiskItem]=Field(description="风险项列表,至少1项。如无明显风险则填写'无显著风险'")suggested_questions:List[InterviewQuestion]=Field(description="建议的面试问题,2-3个")summary:str=Field(description="一句话总结,不超过50字")步骤2:定义简历筛选 Agent
目标:创建专注于结构化评估的 Agent。
# agents.pyfromcrewaiimportAgentdefcreate_resume_screener(llm):"""创建简历筛选 Agent"""returnAgent(role="资深技术招聘专家",goal=("评估候选人简历与'{job_title}'岗位的匹配度,""输出结构化的评估结果(技能匹配度、风险点、面试建议),""每个评估维度必须基于简历中的具体证据。"),backstory=("你在阿里巴巴和字节跳动担任过技术招聘负责人,""累计筛选过超过50000份简历。""你的评估框架包括三个维度:""技能匹配度(硬技能)、经验相关性(项目经验)、""风险信号(频繁跳槽、技能断层、薪资倒挂)。""你坚持一个原则:没有证据的判断就是偏见。""每条评估都必须引用简历中的具体内容作为证据。"),llm=llm,verbose=True,allow_delegation=False,max_iter=10)步骤3:定义 Task(含 Pydantic 输出)
目标:创建使用 Pydantic 输出模型的 Task。
# tasks.pyfromcrewaiimportTaskdefcreate_resume_screening_task(agent,resume_text:str,job_title:str):"""创建结构化简历筛选任务"""returnTask(description=f""" 请评估以下候选人对"{job_title}"岗位的匹配度。 【岗位要求】{job_title}岗位核心要求:相关工作经验3年以上,具备项目独立交付能力。 【候选人简历】{resume_text}【评估要求】 1. 逐项评估技能匹配度,必须引用简历中的证据 2. 识别潜在风险(跳槽频率、技能断档、项目经验夸大信号等) 3. 生成针对性的面试问题 4. 输出决策建议 【输出约束】 - 所有评分必须是0-100的整数 - 风险等级只能是"高/中/低" - 每项评估必须附带简历中的具体证据 """,expected_output=("结构化的候选人评估结果,包含:\n""- 综合评分(0-100)\n""- 面试决策(推荐面试/可考虑/不推荐)\n""- 至少3项技能评估(含匹配度评分和证据)\n""- 亮点列表\n""- 风险列表\n""- 建议面试问题(2-3个)\n""- 一句话总结"),agent=agent,output_pydantic=None,# 将由 crew.py 在运行时设置output_file="outputs/evaluation.md"# 额外保存可读版本)步骤4:组装 Crew 并处理结构化输出
目标:组装 Crew,获取 Pydantic 结构化的评估结果。
# crew.pyfromcrewaiimportCrew,Process,LLMfromagentsimportcreate_resume_screenerfromtasksimportcreate_resume_screening_taskfrommodelsimportCandidateEvaluationimportosimportjsondefscreen_resume(resume_text:str,job_title:str)->CandidateEvaluation:"""筛选单份简历,返回结构化评估结果"""llm=LLM(model=os.getenv("MODEL_NAME","gpt-4o-mini"),temperature=0.2,# 筛选任务偏好确定性输出timeout=120)agent=create_resume_screener(llm)task=create_resume_screening_task(agent,resume_text,job_title)# 关键:设置 Pydantic 输出模型task.output_pydantic=CandidateEvaluation crew=Crew(agents=[agent],tasks=[task],process=Process.sequential,verbose=True)result=crew.kickoff()# 获取 Pydantic 结构化结果ifresult.pydantic:evaluation:CandidateEvaluation=result.pydanticreturnevaluationelse:# 解析失败的兜底处理raiseValueError(f"结构化输出解析失败。原始输出:{result.raw[:500]}")# main.pyfromdotenvimportload_dotenv load_dotenv()fromcrewimportscreen_resumefrommodelsimportCandidateEvaluation,InterviewDecisionimportjsonimportos# 模拟简历数据SAMPLE_RESUME=""" 姓名:张三 工作经历: - 2019-2022 某互联网公司 Python后端开发,负责用户中心微服务架构设计 - 2022-至今 某AI初创公司 高级后端工程师,主导LLM应用平台后端开发 技能:Python(精通)、FastAPI(熟练)、PostgreSQL(熟练)、Docker(熟练)、 Redis(熟练)、K8s(了解)、Go(入门) 项目经验: - 主导设计日活百万的用户中心系统,QPS峰值5000+ - 基于LangChain搭建企业内部知识库问答系统 教育背景:985高校 计算机科学 硕士 其他:GitHub开源项目Star 200+ """defmain():os.makedirs("outputs",exist_ok=True)job_title="高级后端工程师(AI方向)"print(f"正在评估候选人 - 岗位:{job_title}\n")try:evaluation=screen_resume(SAMPLE_RESUME,job_title)# 将Pydantic模型转为字典输出result_dict=evaluation.model_dump()print("="*60)print("【结构化评估结果】")print("="*60)print(json.dumps(result_dict,ensure_ascii=False,indent=2))# 下游系统可直接使用print(f"\n✅ 候选人:{evaluation.candidate_name}")print(f"📊 综合评分:{evaluation.overall_score}/100")print(f"🎯 面试决策:{evaluation.decision.value}")print(f"💡 亮点:{', '.join(evaluation.highlights)}")print(f"⚠️ 风险项数:{len(evaluation.risks)}")# 保存结构化JSON供下游系统消费withopen("outputs/evaluation.json","w",encoding="utf-8")asf:json.dump(result_dict,f,ensure_ascii=False,indent=2)print(f"\n📁 结构化结果已保存至: outputs/evaluation.json")exceptValueErrorase:print(f"❌ 评估失败:{e}")if__name__=="__main__":main()3.4 运行结果
python main.py典型输出:
正在评估候选人 - 岗位: 高级后端工程师(AI方向) [Agent思考] 开始评估候选人张三... ============================================================ 【结构化评估结果】 ============================================================ { "candidate_name": "张三", "overall_score": 78, "decision": "推荐面试", "skill_assessment": [ { "name": "Python", "required": true, "match_level": 90, "evidence": "2019年至今5年Python后端开发经验" }, { "name": "FastAPI", "required": true, "match_level": 85, "evidence": "简历明确列出FastAPI(熟练)" }, { "name": "AI/LLM相关经验", "required": true, "match_level": 70, "evidence": "主导LLM应用平台后端开发,基于LangChain搭建知识库系统" }, { "name": "K8s", "required": false, "match_level": 35, "evidence": "简历标注K8s(了解),深度不足" } ], "highlights": [ "日活百万级用户系统架构经验", "有LLM应用平台实战经验,与岗位AI方向高度契合", "GitHub开源项目有社区认可度" ], "risks": [ { "category": "稳定性", "level": "中", "detail": "当前公司任职2年,上一份工作3年,跳槽频率属正常范围", "mitigation": null }, { "category": "技能缺口", "level": "中", "detail": "K8s和Go仅为入门水平,若岗位需要云原生深度经验则存在缺口", "mitigation": "面试中确认K8s的实际使用深度" } ], "suggested_questions": [ { "question": "你在LLM应用平台项目中具体负责了哪些后端模块?遇到了什么技术挑战?", "purpose": "验证AI方向实际技术深度" }, { "question": "用户中心日活百万时,你们是如何做数据库扩展和高可用设计的?", "purpose": "验证架构能力和高并发经验" } ], "summary": "Python和FastAPI技能扎实,AI方向有实战经验,建议进入技术面进一步验证架构深度。" } ✅ 候选人:张三 📊 综合评分:78/100 🎯 面试决策:推荐面试 💡 亮点:日活百万级用户中心系统架构经验, 有LLM应用平台实战经验, GitHub开源项目有社区认可度 ⚠️ 风险项数:2 📁 结构化结果已保存至: outputs/evaluation.json3.5 输出解析失败的常见原因及解决
| 失败模式 | 症状 | 原因 | 解决 |
|---|---|---|---|
| JSON 格式错误 | 解析抛 JSONDecodeError | 模型输出了非法 JSON | Pydantic 模式比 JSON 稍好——CrewAI 会尝试修复常见格式错误 |
| 类型不匹配 | overall_score为字符串"78分" | 模型忽略了类型约束 | 在 description 中强调"评分必须是纯数字" |
| 必填字段缺失 | risks列表为空但 Pydantic 要求至少1项 | 模型判断"无风险"但未按要求输出 | 在 expected_output 中明确"如无风险,输出’无显著风险’" |
| 枚举值不匹配 | 输出了"建议面试"而非"推荐面试" | 模型用了同义词 | 在 description 中列出枚举的所有合法值 |
3.6 测试验证
# test_structured_output.pyimportpytestfrommodelsimport(CandidateEvaluation,SkillItem,RiskItem,RiskLevel,InterviewDecision,InterviewQuestion)frompydanticimportValidationErrordeftest_valid_evaluation():"""验证合法的评估对象可以成功创建"""eval_data={"candidate_name":"测试候选人","overall_score":75,"decision":InterviewDecision.RECOMMEND,"skill_assessment":[SkillItem(name="Python",required=True,match_level=85,evidence="5年经验")],"highlights":["亮点1"],"risks":[RiskItem(category="稳定性",level=RiskLevel.LOW,detail="无明显风险")],"suggested_questions":[InterviewQuestion(question="测试问题",purpose="测试目的")],"summary":"一句话总结"}evaluation=CandidateEvaluation(**eval_data)assertevaluation.overall_score==75assertevaluation.decision==InterviewDecision.RECOMMENDdeftest_invalid_score_range():"""验证评分超出范围会抛出异常"""withpytest.raises(ValidationError):SkillItem(name="Python",required=True,match_level=150,# 超出0-100evidence="测试")deftest_invalid_risk_level():"""验证非法风险等级会抛出异常"""frompydanticimportValidationError# RiskLevel 是枚举,非法值会抛错withpytest.raises(ValidationError):RiskItem(category="测试",level="超高",detail="测试")deftest_enum_values():"""验证枚举值定义正确"""assertInterviewDecision.RECOMMEND.value=="推荐面试"assertRiskLevel.HIGH.value=="高"assertRiskLevel.LOW.value=="低"deftest_model_can_serialize():"""验证模型可以正确序列化为JSON"""eval_data=CandidateEvaluation(candidate_name="测试",overall_score=80,decision=InterviewDecision.RECOMMEND,skill_assessment=[SkillItem(name="Python",required=True,match_level=90,evidence="测试")],highlights=["亮点"],risks=[RiskItem(category="稳定性",level=RiskLevel.LOW,detail="无风险")],suggested_questions=[InterviewQuestion(question="Q1",purpose="测试")],summary="总结")json_str=eval_data.model_dump_json()assert"测试"injson_strassert"80"injson_strpytest test_structured_output.py-v# 5 passed4. 项目总结
4.1 优点与缺点
| 维度 | Pydantic 输出 | JSON 输出 | Markdown 输出 |
|---|---|---|---|
| 类型安全 | 高,运行时校验 | 中,需手动校验 | 无 |
| 下游消费 | 简单,直接.model_dump() | 简单,json.loads() | 需解析/正则提取 |
| 字段约束 | 强(ge/le/enum/必填) | 无 | 无 |
| 人类可读性 | 低(需工具展示) | 中(可读但不易读) | 高(天然可读) |
| 解析失败率 | 低(CrewAI 会重试) | 中 | 不适用 |
| 给模型"自由度" | 低,强约束可能限制深度 | 中 | 高 |
4.2 适用场景
推荐 Pydantic 输出的场景:
- 需要入库的结构化数据(用户画像、订单摘要、评估报告)
- 下游系统强依赖字段类型和值的范围(计费、风控、合规)
- 批量处理的 AI 产出物(200份简历、1000条用户反馈)
- 前后端分离场景下的 API 响应
- 有自动化测试覆盖的输出校验
不推荐使用的场景:
- 创意类任务(写文案、写报告正文)
- 一次性人工消费的输出(用 Markdown 更合适)
4.3 注意事项
- Pydantic Field 的 description 很重要:它会被传递给 Agent 作为输出约束,写成"技能匹配度(0-100)“而不是"分数”。
- 枚举值要用中文:如果业务语言是中文,枚举值也用中文定义,减少模型"翻译"时的偏差。
- 不要过度嵌套:Pydantic 模型嵌套超过 3 层时,模型的输出准确率显著下降。
- 兜底处理:始终在
result.pydantic为 None 时有降级策略(如用 Markdown 输出 + 人工处理)。
4.4 常见踩坑经验
案例1:模型输出的 JSON 字段名是英文但 Pydantic 定义是中文
现象:Pydantic 解析报"field not found"。
根因:Agent 的 backstory 是英文但 Pydantic 字段是中文,模型在切换语言时"迷路"。
解决:确保 Agent 的 role/goal/backstory 语言与 Pydantic Field description 语言一致。
案例2:必填列表为空导致解析失败
现象:risks字段定义为List[RiskItem],但 Agent 输出[],Pydantic 校验通过但业务不合规。
根因:Pydantic 的List默认允许空列表。
解决:使用Field(min_length=1)约束列表最小长度。
案例3:字符串字段超长导致下游截断
现象:summary字段 Agent 写了 500 字,下游展示时被截断。
根因:没有在 Pydantic Field 中限制字符串最大长度。
解决:使用Field(max_length=100)限制长度。
4.5 思考题
如果一份简历需要多个 Agent 协同评估(技术评估 + 文化匹配 + 薪资匹配),你是让每个 Agent 输出独立的 Pydantic 模型,还是让最后一个 Agent 汇总成一个大模型?两种方案在维护性、可靠性和扩展性上的差异是什么?
当 Pydantic 解析失败时,CrewAI 会触发重试。但如果连续 3 次重试都失败(比如模型就是输出不了合法的枚举值),你该如何设计降级方案?是切换为更宽松的 JSON 模式,还是直接人工接管?
答案将在后续章节揭晓。
延伸阅读与资源
10倍开发者的 Dify 魔法书:从零构建全栈 AI 应用
后端工程师转型AI第一课-Ollama 与私有化大模型实战
大型语言模型(LLM) vLLM 高性能推理落地实战
Agent开发之LlamaIndex 实战修炼与源码进阶
大语言模型Transformers 实战修炼与源码剖析