结构化输出

演示如何引导 LLM 输出结构化 JSON：构建带 schema 的 prompt、从模型回复中 稳健地提取 JSON（含 json 代码块包裹的情况），并校验关键字段。 解析逻辑全部基于标准库，便于离线测试。 对应教程：第4卷《提示工程》第3章 结构化提示。

文件路径: examples/volume-04-prompt-engineering\structured_output.py

---

源代码

"""结构化输出示例。

演示如何引导 LLM 输出结构化 JSON：构建带 schema 的 prompt、从模型回复中
稳健地提取 JSON（含 ```json``` 代码块包裹的情况），并校验关键字段。
解析逻辑全部基于标准库，便于离线测试。

对应教程：第4卷《提示工程》第3章 结构化提示。
"""

from __future__ import annotations

import json
import re
from dataclasses import dataclass, field
from typing import Any


@dataclass
class JsonSchemaField:
    """JSON Schema 字段定义。

    Attributes:
        name: 字段名。
        type: 字段类型（如 ``string``、``integer``）。
        description: 字段描述。
        required: 是否必填。
        enum: 可选的枚举值列表。
    """

    name: str
    type: str
    description: str
    required: bool = True
    enum: list[str] | None = None


@dataclass
class StructuredPromptBuilder:
    """结构化输出 prompt 构建器。

    Attributes:
        task: 任务说明。
        fields: 期望输出的字段定义列表。
    """

    task: str
    fields: list[JsonSchemaField] = field(default_factory=list)

    def add_field(self, f: JsonSchemaField) -> StructuredPromptBuilder:
        """添加一个字段定义。

        Args:
            f: 字段定义。

        Returns:
            ``self``。
        """
        self.fields.append(f)
        return self

    def build_schema(self) -> dict[str, Any]:
        """构建 JSON Schema 字典。

        Returns:
            描述输出结构的 JSON Schema。
        """
        properties: dict[str, Any] = {}
        required: list[str] = []
        for f in self.fields:
            prop: dict[str, Any] = {"type": f.type, "description": f.description}
            if f.enum is not None:
                prop["enum"] = f.enum
            properties[f.name] = prop
            if f.required:
                required.append(f.name)
        return {
            "type": "object",
            "properties": properties,
            "required": required,
        }

    def build_prompt(self, user_request: str) -> str:
        """构建要求模型输出结构化 JSON 的 prompt。

        Args:
            user_request: 用户请求。

        Returns:
            完整 prompt 字符串。
        """
        schema = self.build_schema()
        return (
            f"{self.task}\n\n"
            f"请严格按以下 JSON Schema 输出，不要包含任何额外说明：\n"
            f"{json.dumps(schema, ensure_ascii=False, indent=2)}\n\n"
            f"用户请求：{user_request}"
        )


_JSON_BLOCK_RE = re.compile(r"```(?:json)?\s*(\{.*?\}|\[.*?\])\s*```", re.DOTALL)
_BARE_JSON_RE = re.compile(r"(\{.*\}|\[.*\])", re.DOTALL)


def extract_json(text: str) -> Any:
    """从模型回复中提取 JSON 对象或数组。

    依次尝试：
        1. 整段文本直接 ``json.loads``；
        2. `` ```json ... ``` `` 代码块；
        3. 第一个 ``{...}`` 或 ``[...]`` 片段。

    Args:
        text: 模型回复文本。

    Returns:
        解析后的 Python 对象。

    Raises:
        ValueError: 无法提取合法 JSON 时。
    """
    text = text.strip()
    # 1. 直接解析
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        pass
    # 2. 代码块
    m = _JSON_BLOCK_RE.search(text)
    if m:
        try:
            return json.loads(m.group(1))
        except json.JSONDecodeError:
            pass
    # 3. 裸 JSON 片段（取第一个）
    m = _BARE_JSON_RE.search(text)
    if m:
        try:
            return json.loads(m.group(1))
        except json.JSONDecodeError:
            pass
    raise ValueError(f"无法从文本中提取 JSON: {text[:80]!r}")


def validate_required_fields(data: Any, required: list[str]) -> list[str]:
    """校验字典是否包含全部必填字段。

    Args:
        data: 待校验对象（应为 dict）。
        required: 必填字段名列表。

    Returns:
        缺失字段名列表；为空表示全部通过。
    """
    if not isinstance(data, dict):
        return list(required)
    return [f for f in required if f not in data]


def main() -> None:
    """演示结构化输出 prompt 构建与 JSON 提取。"""
    builder = StructuredPromptBuilder(task="从用户评论中抽取结构化信息。")
    builder.add_field(
        JsonSchemaField("sentiment", "string", "情感倾向", enum=["正面", "负面", "中性"])
    )
    builder.add_field(JsonSchemaField("keywords", "array", "关键词列表"))
    builder.add_field(JsonSchemaField("score", "integer", "1-5 分评分"))

    print(builder.build_prompt("这款 Agent 框架文档清晰、性能优秀，强烈推荐。"))

    print("\n--- JSON 提取 ---")
    samples = [
        '{"sentiment": "正面", "score": 5}',
        '```json\n{"sentiment": "中性", "score": 3}\n```',
        '好的，结果如下：{"sentiment": "负面", "score": 2} 希望有帮助。',
    ]
    for s in samples:
        data = extract_json(s)
        missing = validate_required_fields(data, ["sentiment", "score"])
        print(f"  提取: {data}  缺失字段: {missing}")


if __name__ == "__main__":
    main()