Day 15 / 共 60 天 · 阶段3 Task 深入

结构化输出：把 LLM 吐的文本"翻译"成对象

昨天 TaskOutput 有三副面孔（raw/pydantic/json_dict）。但 pydantic 和 json_dict 是怎么从一段文本变出来的？今天钻进去看：output_pydantic、output_json、response_model 三个字段的分工，产出在 _execute_core 里怎么按类型分流，以及核心的 convert_to_model 如何把 LLM 那段"可能不太干净的 JSON 文本"解析成合法对象——包括解析失败时的多级兜底。源码在 task.py 与 utilities/converter.py。

📍 你在 60 天里的位置（阶段3：Task 深入 · 共 6 天）

阶段2 Agent→ D13 Task 模型→ D14 输出与护栏→ D15 结构化输出→ D16 context 依赖→ D17 异步任务→ D18 条件任务→ 阶段4 Crew

L01

痛点：拿到一段文本，程序没法直接用

🤔 痛点LLM 返回的是文本。你想在代码里 result.risks[0].level 这样用，就得先把那段文本解析成结构。但 LLM 的"JSON"经常不干净：外面裹着 ```json ``` 代码块、前面加句"好的，这是结果"、尾巴多个逗号……直接 json.loads 十有八九报错。每个任务都手写一套"提取+清洗+解析+校验"太痛苦。

💡 一句话本质：声明一个模型，框架负责把文本"塞"进去你只要给任务一个 Pydantic 模型（output_pydantic=Report），CrewAI 就负责：引导 LLM 按这个模型的结构输出 → 拿到文本 → 尽力解析成 JSON → 用模型校验 → 给你一个类型安全的对象。"要什么结构"你声明，"怎么把文本变成它"框架包办。

🧩 生活类比：填表 vs 写作文让 LLM 自由发挥像"写作文"（raw 文本，你难提取信息）；给它一个模型像"发一张固定格子的表格"让它填。填完的表格，程序能精确读每个格子（report.risks）。output_pydantic 就是那张表格的定义。

L02

三个字段的分工

Task 上有三个和结构化输出相关的字段（task.py:169、task.py:179、task.py:189）：

# task.py:169
output_json: ... = Field(
    description="A Pydantic model to be used to create a JSON output.", default=None)
# task.py:179
output_pydantic: ... = Field(
    description="A Pydantic model to be used to create a Pydantic output.", default=None)
# task.py:189
response_model: ... = Field(
    description="A Pydantic model for structured LLM outputs using native provider features.",
    default=None)

图注：output_pydantic / output_json 是"生成文本后再解析"；response_model 走 provider 原生结构化输出，从源头保证结构。

💡 设计取舍①：response_model（原生）vs output_pydantic（事后转换） output_pydantic：不挑模型，任何 LLM 都能用——因为它是"生成完普通文本，再由框架解析+校验"。缺点是解析可能失败（文本不规范），要靠兜底。response_model：让 provider（如 OpenAI 的 structured outputs）在生成时就强制按 schema 出结构，几乎不会解析失败。缺点是依赖 provider 支持这个特性。所以 CrewAI 两条路都留着：能用原生就用原生（稳），不支持就退回事后转换（通用）。这是"追求最优、但保证兜底"的典型。

L03

产出分流：_execute_core 里的三条路

回到 D13 看过的 _execute_core，Agent 返回 result 后有一段关键的类型分流（task.py:798）：

# task.py:798
if isinstance(result, BaseModel):        # ① Agent 已经吐了对象（response_model 情形）
    raw = result.model_dump_json()
    if self.output_pydantic:
        pydantic_output = result; json_output = None
    elif self.output_json:
        pydantic_output = None; json_output = result.model_dump()
    else:
        pydantic_output = None; json_output = None
elif not self._guardrails and not self._guardrail:  # ② 是文本，且没护栏 → 立刻解析
    raw = result
    pydantic_output, json_output = self._export_output(result)
else:                                     # ③ 是文本，但有护栏 → 先不解析，等护栏放行再说
    raw = result
    pydantic_output, json_output = None, None

① result 已是 BaseModel说明底层用了原生结构化（response_model）：直接拿对象，按 output_pydantic/output_json 决定填哪副面孔。raw 用 model_dump_json() 补上文本面孔。

② 文本 + 无护栏正常情况：立刻调 _export_output 把文本解析成结构（L04）。

③ 文本 + 有护栏顺序讲究：先不解析（pydantic/json 都置 None），因为护栏（D14）可能会改写 raw；等护栏放行、拿到最终文本后再解析（回看 D14 的 _export_output 调用）。避免"解析了又被护栏改、白解析一次"。

💡 为什么护栏在场时要延后解析？护栏可能返回一个修正后的字符串（D14 L06），那这段修正后的文本才是"最终产出"。若在护栏前就解析，护栏一改写，解析结果就作废了。所以框架把解析推迟到护栏之后——这是控制流顺序上的一个精心安排。

L04

_export_output：解析的入口

把文本变成结构，入口是 _export_output（task.py:1116），它调 convert_to_model 拿结果，再用 _unpack_model_output 拆成 (pydantic, json) 两格：

# task.py:1116
def _export_output(self, result) -> tuple[BaseModel | None, dict | None]:
    pydantic_output, json_output = None, None
    if self.output_pydantic or self.output_json:            # 只有声明了模型才解析
        model_output = convert_to_model(
            result, self.output_pydantic, self.output_json, self.agent, self.converter_cls)
        pydantic_output, json_output = self._unpack_model_output(model_output)
    return pydantic_output, json_output

# task.py:1153
@staticmethod
def _unpack_model_output(model_output) -> tuple[BaseModel | None, dict | None]:
    if isinstance(model_output, BaseModel):  return model_output, None   # 对象
    if isinstance(model_output, dict):       return None, model_output   # 字典
    if isinstance(model_output, str):                                    # 还是字符串
        try:    return None, json.loads(model_output)
        except json.JSONDecodeError:  return None, None                  # 彻底失败：都 None
    return None, None

if output_pydantic or output_json没声明模型就不解析（直接返回两个 None）——纯文本任务不做无谓转换。

convert_to_model(...)核心解析器（L05），返回可能是对象、字典或原样字符串。

_unpack_model_output按返回类型分拣：对象放 pydantic 格、字典放 json 格、还是字符串就再试一次 json.loads，实在不行两格都 None（宽容降级，不炸）。

💡 _get_output_format（task.py:1168）根据你设了哪个字段返回 JSON / PYDANTIC / RAW，这就是 D14 里 TaskOutput 那张"用哪副面孔"标签的来源。

L05

convert_to_model：解析的主干与降级

真正的解析主干在 convert_to_model（utilities/converter.py:190）。它的结构就是一层层"试着解析，失败就换更宽容的方式"：

# utilities/converter.py:190
def convert_to_model(result, output_pydantic, output_json, agent, converter_cls=None):
    model = output_pydantic or output_json
    if model is None:
        return result                                   # 没模型，原样返回

    if isinstance(result, BaseModel):                   # 已经是对象
        if isinstance(result, model):
            return result.model_dump() if output_json else result
        result = result.model_dump_json()

    if converter_cls:                                   # 用户自定义转换器优先
        return convert_with_instructions(...)

    try:
        escaped_result = json.dumps(json.loads(result, strict=False))  # ★先规整一遍 JSON
        return validate_model(escaped_result, model, is_json_output=bool(output_json))
    except json.JSONDecodeError:
        return handle_partial_json(...)                 # 不是合法 JSON → 部分兜底（L06）
    except ValidationError:
        return handle_partial_json(...)                 # 结构对不上 → 也走兜底
    except Exception as e:
        ... ; return result                             # 其它意外 → 保底返回原文

图注：一条"从严格到宽容"的降级链——直接解析 → 正则抠 JSON → 回喂 LLM 重格式化 → 保底返回原文，尽最大努力不让你两手空空。

💡 设计取舍②：为什么先 json.dumps(json.loads(result)) 绕一圈？你可能觉得直接 model.model_validate_json(result) 就行。但 LLM 的 JSON 常有小瑕疵：单引号、尾逗号、多余空白。json.loads(result, strict=False) 用宽松模式先把它读成 Python 对象，再 json.dumps 序列化回一段规范的 JSON 文本，然后才交给 Pydantic 校验。这一"读进来再吐回去"等于洗了一遍澡，把格式毛刺磨平，让后续校验更稳。代价是多一次序列化开销，但比"直接校验失败"划算得多。

L06

部分 JSON 兜底：从一堆废话里抠出 JSON

如果整段不是合法 JSON（比如 LLM 说了"好的，结果如下：{...}还有什么需要吗"），就轮到 handle_partial_json（utilities/converter.py:280）用正则把中间那段 JSON 抠出来：

# utilities/converter.py:280
def handle_partial_json(result, model, is_json_output, agent, converter_cls=None):
    match = _JSON_PATTERN.search(result)          # ★正则找出文本里 {...} 那一段
    if match:
        try:
            parsed = json.loads(match.group(), strict=False)
        except json.JSONDecodeError:
            return convert_with_instructions(...)  # 抠出来还不合法 → 回喂 LLM 重整
        try:
            exported_result = model.model_validate(parsed)   # 校验
            return exported_result.model_dump() if is_json_output else exported_result
        except ValidationError:
            raise
        except Exception as e:
            ...

_JSON_PATTERN.search用正则在整段文本里找 { ... } 这样的 JSON 片段——哪怕前后裹着大段废话也能抠出来。

抠出来仍非法再降一级：convert_with_instructions——把模型 schema 当"指令"回喂给 LLM，让它重新按格式输出一次（要花一次 LLM 调用）。

model.model_validate(parsed)抠出来的 JSON 用模型校验：字段/类型对得上就成对象，对不上抛 ValidationError。

📝 真实值走一遍 LLM 返回 "当然！这是分析结果：\n```json\n{\"risks\": [\"汇率\"]}\n```\n希望有帮助"。
主干 json.loads 失败（整段不是 JSON）→ handle_partial_json 正则抠出 {"risks": ["汇率"]} → model_validate 成 Report(risks=["汇率"])。那堆客套话被自动无视了。

L07

互斥边界 + 关键取舍

回顾 D13 见过的互斥校验（task.py:548）——它保证 output_json 和 output_pydantic 不能同时设：

# task.py:548
@model_validator(mode="after")
def check_output(self) -> Self:
    output_types = [self.output_json, self.output_pydantic]
    if len([type for type in output_types if type]) > 1:
        raise PydanticCustomError(
            "output_type",
            "Only one output type can be set, either output_pydantic or output_json.", {})
    return self

⚠️ 边界①：解析失败不报错，而是"降级返回原文" 看 convert_to_model 最后的 except Exception: return result——万一所有解析路径都失败，框架不抛异常中断任务，而是把原始文本原样返回。为什么这里宽容（对比 D13/D14 的 fail-fast）？因为结构化解析失败不是"配置矛盾"，而是"LLM 这次没配合"——任务的核心产出（raw 文本）其实拿到了，只是结构没解析出来。与其让整个 crew 崩掉，不如保住 raw、pydantic/json 置空，让你至少拿到文本。矛盾配置要 fail-fast，尽力而为的解析要优雅降级——两种"边界哲学"用在不同地方。

⚠️ 边界②：output_json 拿到的是 dict，不是 JSON 字符串名字容易误导：output_json 最终填的是 json_dict（一个 Python 字典），不是字符串。想要字符串用 output.json 属性（D14，且会检查 format）。而 output_pydantic 填的才是模型对象。选哪个取决于下游想要 dict 还是带方法的对象。

👶 小白：那我到底该用 output_pydantic、output_json 还是 response_model？

👨‍🏫 老师：默认首选 output_pydantic——拿到带类型和方法的对象，最好用，且任何模型都能用。若下游只想要普通字典（比如直接塞进另一个 API），用 output_json。若你的 provider 支持原生结构化输出、且你想把"解析失败"的概率降到最低，用 response_model（从源头让 LLM 吐结构）。三者的解析/校验最终都殊途同归到 TaskOutput 的三副面孔上。

L08

今日小结 + 动手

                🧠 今天你应该能回答
                结构化输出解决什么痛点？为什么"填表"胜过"写作文"？
output_pydantic / output_json / response_model 三者的分工？
为什么 response_model（原生）比 output_pydantic（事后转换）更不容易解析失败？
产出在 _execute_core 里为什么分三条路？护栏在场时为什么延后解析？
convert_to_model 的降级链有哪几级？
为什么解析前要 json.dumps(json.loads(...)) 绕一圈？
解析彻底失败时框架为什么不报错而是返回原文？

            

✋ 10 分钟动手

P=lib/crewai/src/crewai
# 1. 三个字段声明
sed -n '169,198p' $P/task.py

# 2. 产出分流三条路
sed -n '798,814p' $P/task.py

# 3. _export_output + 拆包
sed -n '1116,1173p' $P/task.py

# 4. convert_to_model 主干 + 部分 JSON 兜底
sed -n '190,258p' $P/utilities/converter.py
sed -n '280,320p' $P/utilities/converter.py

明日预告 · Day 16：任务不是孤岛——第二个任务常要吃第一个任务的产出。明天讲 context 任务依赖：那个神秘的 NOT_SPECIFIED 三态到底怎么工作、_get_context 如何把上游产出拼成上下文喂给下游、以及 Crew 怎么校验"不许依赖未来的任务"。

← Day 14 输出与护栏 Day 16 · context 依赖 →