Day 4：结构化输出与第一个阶段验收

文章目录

• • 一、本篇目标
  • 二、核心概念：为什么需要结构化输出
  • • 1. 自然语言输出的问题
    • 2. 结构化输出的优势
    • 3. 什么时候需要结构化输出
  • 三、深入理解：让模型输出 JSON 的三种方法
  • • 方法 1：在提示词中明确要求
    • 方法 2：使用 JSON Mode
    • 方法 3：使用 Function Calling（推荐）
  • 四、案例分析：常见的结构化输出失败
  • • 1. 案例 1：输出包含额外文字
    • 2. 案例 2：字段缺失或类型错误
    • 3. 案例 3：枚举值不符合约束
    • 4. 案例 4：JSON 格式不合法
  • 五、项目需求：为 WorkMate 增加结构化输出
  • 六、准备开发环境
  • 七、实现带结构化输出的 Agent
  • • 这段代码真正做了什么
  • 八、运行效果
  • 九、常见问题与排错
  • • 1. 模型仍然输出额外文字
    • 2. JSON 解析失败
    • 3. 字段缺失或类型错误
    • 4. 枚举值不符合约束
    • 5. 为什么不用 Pydantic 或其他验证库
  • 十、工程化改进：现在还不能上线
  • 十一、本篇小结
  • 十二、课后练习
  • • 练习 1：增加一个任务列表功能
    • 练习 2：实现批量任务分析
    • 练习 3：增加优先级冲突检测
    • 阶段验收

✍创作者：全栈弄潮儿 🏡 个人主页：全栈弄潮儿的个人主页 🏙️ 个人社区，欢迎你的加入：全栈开发社区 📙 专栏：AI Agent 开发实战：从 0 到生产级智能体

这是《AI Agent 开发实战：从 0 到生产级智能体》的第 4 篇，也是第一阶段的收官篇。

前三篇我们搭建了 Agent 骨架，实现了多轮对话和上下文管理。但你可能已经发现一个问题：Agent 的回答格式不稳定。有时候它输出 JSON，有时候输出纯文本，有时候格式对但字段缺失。

这在聊天场景下可以接受，但在真实业务中往往不够用。当你需要把 Agent 的输出存进数据库、传给下游系统、或者作为工具调用的参数时，自然语言输出就变成了一个定时炸弹。

本篇要解决这个问题。我们会学习如何让 Agent 稳定输出结构化数据，并在最后完成第一阶段的验收。

完成后，你会得到一个能够稳定输出 JSON 的 Agent，它能够：

根据任务类型输出标准 JSON 格式
↓
自动校验输出是否符合数据模型
↓
格式错误时自动重试或修正
↓
为后续 Tool Calling 准备好数据结构

一、本篇目标

完成下面 5 件事：

本篇的最终验收标准是：

[ ] Agent 能够根据任务类型输出标准 JSON。
[ ] 输出格式错误时能够自动重试或修正。
[ ] 数据模型校验通过后才返回结果。
[ ] 完成"智能任务助手 V1"的全部功能验收。
[ ] 能够解释结构化输出在后续章节中的作用。

二、核心概念：为什么需要结构化输出

1. 自然语言输出的问题

前三篇的 Agent 输出都是自然语言：

用户：帮我分析一下这个任务的优先级

Agent：根据你提供的信息，这个任务的优先级应该是"高"。原因是截止日期比较紧，而且涉及到核心业务功能。建议你优先处理。

这种输出对人类友好，但对程序不友好。如果你想把"优先级"这个结果存进数据库，或者传给下一个工具，你需要从这段文字里提取信息——这本身又是一个 AI 任务。

2. 结构化输出的优势

如果 Agent 输出的是标准 JSON：

{
"task_id": "task_001",
"priority": "high",
"reason": "截止日期紧，涉及核心业务",
"suggestion": "优先处理"
}

下游系统可以直接解析和使用，不需要额外的提取步骤。

3. 什么时候需要结构化输出

简单判断：如果输出需要被程序消费，就应该结构化。

三、深入理解：让模型输出 JSON 的三种方法

方法 1：在提示词中明确要求

最直接的方式是在系统指令中明确要求输出 JSON：

system_prompt = """你是一个任务分析助手。

对于每个任务，你必须输出以下 JSON 格式：

{
"task_id": "任务ID",
"priority": "high/medium/low",
"reason": "优先级判断理由",
"suggestion": "处理建议"
}

注意：
– 只输出 JSON，不要输出其他内容
– priority 只能是 high、medium 或 low
– 所有字段都必须填写"""

优点：简单直接，不需要额外代码。

缺点：模型有时会忽略要求，输出额外的解释文字，或者格式不完全符合。

方法 2：使用 JSON Mode

部分模型支持 JSON Mode，强制输出合法的 JSON：

response = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
response_format={"type": "json_object"}
)

优点：模型保证输出合法 JSON，不需要额外校验。

缺点：不是所有模型都支持；只能保证 JSON 合法，不能保证字段符合预期。

方法 3：使用 Function Calling（推荐）

通过定义函数参数，让模型输出符合 Schema 的结构化数据：

tools = [{
"type": "function",
"function": {
"name": "analyze_task",
"description": "分析任务优先级",
"parameters": {
"type": "object",
"properties": {
"task_id": {"type": "string"},
"priority": {"type": "string", "enum": ["high", "medium", "low"]},
"reason": {"type": "string"},
"suggestion": {"type": "string"}
},
"required": ["task_id", "priority", "reason", "suggestion"]
}
}
}]

response = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
tools=tools,
tool_choice={"type": "function", "function": {"name": "analyze_task"}}
)

优点：模型输出严格符合 Schema，不需要额外校验；支持枚举类型约束。

缺点：需要模型支持 Function Calling；代码稍复杂。

本篇先实现方法 1 和方法 2，方法 3 留到下一篇 Tool Calling 详细讲解。

四、案例分析：常见的结构化输出失败

1. 案例 1：输出包含额外文字

问题：要求输出 JSON，但模型输出了：

根据分析，结果如下：

{
"priority": "high"
}

建议你优先处理。

原因：提示词中没有强调"只输出 JSON"。

解决：

system_prompt = """…

【重要】只输出 JSON，不要输出任何其他文字。不要有前言、解释或总结。"""

2. 案例 2：字段缺失或类型错误

问题：要求输出 4 个字段，但模型只输出了 3 个：

{
"task_id": "task_001",
"priority": "high",
"reason": "截止日期紧"
}

缺少 suggestion 字段。

原因：提示词中没有强调"所有字段都必须填写"。

解决：

system_prompt = """…

注意：
– 所有字段都必须填写，不能省略
– 如果某个字段没有信息，填写"无"或空字符串"""

3. 案例 3：枚举值不符合约束

问题：要求 priority 只能是 high/medium/low，但模型输出了：

{
"priority": "urgent"
}

原因：模型没有严格遵守枚举约束。

解决：

VALID_PRIORITIES = {"high", "medium", "low"}

def validate_priority(priority: str) –> str:
if priority.lower() in VALID_PRIORITIES:
return priority.lower()
# 尝试映射常见变体
mapping = {
"urgent": "high",
"critical": "high",
"normal": "medium",
"low": "low"
}
return mapping.get(priority.lower(), "medium")

4. 案例 4：JSON 格式不合法

问题：模型输出了不合法的 JSON：

{
"priority": "high",
"reason": "截止日期紧，需要加班"
}

注意 reason 字段末尾缺少引号。

原因：模型生成时偶尔会出现格式错误。

解决：

import json

def parse_json_with_retry(text: str, max_retries: int = 3) –> dict:
"""尝试解析 JSON，失败时重试"""
for attempt in range(max_retries):
try:
return json.loads(text)
except json.JSONDecodeError:
if attempt < max_retries – 1:
# 可以尝试修复常见问题，如缺少引号、多余逗号等
text = fix_common_json_errors(text)
else:
raise ValueError(f"JSON 解析失败，已重试 {max_retries} 次")

五、项目需求：为 WorkMate 增加结构化输出

为了让 Agent 能够稳定输出结构化数据，我们需要：

• 实现一个任务分析功能，输出标准 JSON。
• 增加数据模型校验，确保字段完整、类型正确。
• 实现格式错误时的重试机制。
• 支持 JSON Mode（如果模型支持）。
• 完成第一阶段验收，交付"智能任务助手 V1"。

暂时不做：

• 不实现复杂的 Schema 验证库（如 Pydantic）。
• 不实现 Function Calling（下一篇实现）。
• 不实现多任务批量处理。

这几个限制很重要。我们要先验证"提示词 + 校验 + 重试"能否稳定工作，再引入更复杂的机制。

六、准备开发环境

本篇复用前三篇的项目环境。如果你已经完成了前面的章节，可以直接使用 agent-workbench 项目。

如果还没有，请先完成第 1 篇的环境搭建：

mkdir agent-workbench
cd agent-workbench
python -m venv .venv
source .venv/bin/activate # macOS / Linux
python -m pip install -r requirements.txt

确保 .env 中已经配置了有效的 API Key。

七、实现带结构化输出的 Agent

创建 agent.py：

from __future__ import annotations

import json
import os
from typing import Any

from dotenv import load_dotenv
from openai import OpenAI

SYSTEM_PROMPT = """你是 WorkMate，一个专业的任务分析助手。

你的职责是分析用户提供的任务信息，并输出结构化的分析结果。

对于每个任务，你必须输出以下 JSON 格式：

{
"task_id": "任务ID（如果用户没有提供，生成一个唯一ID）",
"title": "任务标题",
"priority": "high/medium/low",
"deadline": "截止日期（如果用户没有提供，填写"未指定"）",
"reason": "优先级判断理由",
"suggestion": "处理建议"
}

【重要】
– 只输出 JSON，不要输出任何其他文字
– 不要有前言、解释或总结
– 所有字段都必须填写，不能省略
– priority 只能是 high、medium 或 low
– 如果某个字段没有信息，填写"未指定"或空字符串"""

VALID_PRIORITIES = {"high", "medium", "low"}

def validate_and_fix_task(data: dict) –> dict:
"""校验并修正任务数据"""
# 确保所有必需字段存在
required_fields = ["task_id", "title", "priority", "deadline", "reason", "suggestion"]
for field in required_fields:
if field not in data:
data[field] = "未指定"

# 校验并修正 priority
priority = data.get("priority", "").lower()
if priority not in VALID_PRIORITIES:
# 尝试映射常见变体
mapping = {
"urgent": "high",
"critical": "high",
"normal": "medium",
"standard": "medium",
"low": "low",
"trivial": "low"
}
data["priority"] = mapping.get(priority, "medium")

return data

def parse_json_with_retry(text: str, max_retries: int = 3) –> dict:
"""尝试解析 JSON，失败时重试"""
# 尝试提取 JSON 部分
text = text.strip()

# 如果包含 markdown 代码块，提取其中的 JSON
if "```json" in text:
start = text.find("```json") + 7
end = text.find("```", start)
text = text[start:end].strip()
elif "```" in text:
start = text.find("```") + 3
end = text.find("```", start)
text = text[start:end].strip()

for attempt in range(max_retries):
try:
return json.loads(text)
except json.JSONDecodeError as e:
if attempt < max_retries – 1:
# 尝试修复常见问题
text = fix_common_json_errors(text)
else:
raise ValueError(f"JSON 解析失败：{e}")

return {}

def fix_common_json_errors(text: str) –> str:
"""修复常见的 JSON 格式错误"""
# 移除末尾多余的逗号
text = text.replace(",}", "}").replace(",]", "]")

# 尝试补全缺失的引号（简单启发式）
# 这里只是示例，实际场景可能需要更复杂的修复逻辑

return text

class Agent:
"""带结构化输出的 Agent"""

def __init__(
self,
client: OpenAI,
model: str,
use_json_mode: bool = False,
) –> None:
self.client = client
self.model = model
self.use_json_mode = use_json_mode
self.messages: list[dict[str, str]] = [
{"role": "system", "content": SYSTEM_PROMPT}
]

def reset(self) –> None:
"""清空当前会话，但保留系统指令。"""
self.messages = [{"role": "system", "content": SYSTEM_PROMPT}]

def analyze_task(self, user_input: str) –> dict:
"""分析任务并返回结构化结果"""
self.messages.append({"role": "user", "content": user_input})

# 构建请求参数
request_params = {
"model": self.model,
"messages": self.messages,
"temperature": 0.1, # 降低随机性，提高格式稳定性
}

# 如果启用 JSON Mode
if self.use_json_mode:
request_params["response_format"] = {"type": "json_object"}

response = self.client.chat.completions.create(**request_params)

answer = response.choices[0].message.content
if not answer:
raise RuntimeError("模型返回了空内容")

self.messages.append({"role": "assistant", "content": answer})

# 解析 JSON
try:
data = parse_json_with_retry(answer)
# 校验并修正
data = validate_and_fix_task(data)
return data
except Exception as e:
# 如果解析失败，返回错误信息
return {
"task_id": "error",
"title": "解析失败",
"priority": "medium",
"deadline": "未指定",
"reason": f"JSON 解析失败：{e}",
"suggestion": "请重新描述任务信息"
}

def chat(self, user_input: str) –> str:
"""普通对话（非结构化输出）"""
self.messages.append({"role": "user", "content": user_input})

response = self.client.chat.completions.create(
model=self.model,
messages=self.messages,
temperature=0.2,
)

answer = response.choices[0].message.content
if not answer:
raise RuntimeError("模型返回了空内容")

self.messages.append({"role": "assistant", "content": answer})
return answer

def build_agent() –> Agent:
load_dotenv()

api_key = os.getenv("OPENAI_API_KEY")
if not api_key or api_key == "replace-with-your-key":
raise RuntimeError(
"没有找到有效的 OPENAI_API_KEY，请先在 .env 中配置 API Key。"
)

client_options: dict[str, Any] = {"api_key": api_key}
base_url = os.getenv("OPENAI_BASE_URL")
if base_url:
client_options["base_url"] = base_url

client = OpenAI(**client_options)
model = os.getenv("OPENAI_MODEL", "gpt-4o-mini")
use_json_mode = os.getenv("USE_JSON_MODE", "false").lower() == "true"
return Agent(client=client, model=model, use_json_mode=use_json_mode)

def main() –> None:
try:
agent = build_agent()
except Exception as exc:
print(f"启动失败：{exc}")
return

print("Agent 已启动。")
print("输入任务信息进行分析，输入 /chat 切换到普通对话模式。")
print("输入 /reset 清空上下文，输入 /exit 退出。")

mode = "analyze" # analyze 或 chat

while True:
try:
user_input = input("\\n你：").strip()
except (EOFError, KeyboardInterrupt):
print("\\nAgent 已退出。")
return

if not user_input:
continue
if user_input == "/exit":
print("Agent 已退出。")
return
if user_input == "/reset":
agent.reset()
print("会话上下文已清空。")
continue
if user_input == "/chat":
mode = "chat" if mode == "analyze" else "analyze"
print(f"已切换到{'普通对话' if mode == 'chat' else '任务分析'}模式。")
continue

try:
if mode == "analyze":
result = agent.analyze_task(user_input)
print("\\nAgent：")
print(json.dumps(result, ensure_ascii=False, indent=2))
else:
print(f"\\nAgent：{agent.chat(user_input)}")
except Exception as exc:
print(f"调用失败：{exc}")

if __name__ == "__main__":
main()

这段代码真正做了什么

代码实现了结构化输出的核心能力：

Agent 现在能够稳定输出 JSON，并在格式错误时自动修复或重试。

八、运行效果

启动程序：

python agent.py

一次可能的运行过程如下：

Agent 已启动。
输入任务信息进行分析，输入 /chat 切换到普通对话模式。
输入 /reset 清空上下文，输入 /exit 退出。

你：我有一个任务，需要在明天之前完成用户认证模块的开发

Agent：
{
"task_id": "task_001",
"title": "用户认证模块开发",
"priority": "high",
"deadline": "明天",
"reason": "截止日期紧迫，涉及核心功能",
"suggestion": "立即开始开发，优先完成核心逻辑"
}

你：还有一个任务，下周整理一下文档

Agent：
{
"task_id": "task_002",
"title": "文档整理",
"priority": "low",
"deadline": "下周",
"reason": "时间充裕，非紧急任务",
"suggestion": "可以安排在空闲时间处理"
}

你：/chat
已切换到普通对话模式。

你：刚才分析的两个任务，哪个更紧急？

Agent：第一个任务（用户认证模块开发）更紧急。它的截止日期是明天，优先级为 high；而第二个任务（文档整理）的截止日期是下周，优先级为 low。建议优先完成用户认证模块。

你：/reset
会话上下文已清空。

这里要重点观察几个现象：

九、常见问题与排错

1. 模型仍然输出额外文字

检查提示词中是否明确写了"只输出 JSON"。有时候模型会忽略隐含的要求，需要显式强调：

system_prompt = """…

【重要】只输出 JSON，不要输出任何其他文字。不要有前言、解释或总结。"""

2. JSON 解析失败

可能的原因：

• 模型输出了不合法的 JSON（如缺少引号、多余逗号）。
• 模型输出了 markdown 代码块，但没有正确提取。

解决方案：

• 使用 parse_json_with_retry() 自动提取和修复。
• 启用 JSON Mode（如果模型支持）。
• 在提示词中给出完整的 JSON 示例。

3. 字段缺失或类型错误

检查提示词中是否明确写了"所有字段都必须填写"。如果模型仍然省略字段，可以在代码中增加默认值：

def validate_and_fix_task(data: dict) –> dict:
required_fields = ["task_id", "title", "priority", "deadline", "reason", "suggestion"]
for field in required_fields:
if field not in data:
data[field] = "未指定"
return data

4. 枚举值不符合约束

模型有时会输出不在枚举范围内的值（如 urgent 而不是 high）。解决方案：

• 在提示词中明确列出所有可选值。
• 在代码中增加映射逻辑，把常见变体映射到标准值。

VALID_PRIORITIES = {"high", "medium", "low"}

def validate_priority(priority: str) –> str:
if priority.lower() in VALID_PRIORITIES:
return priority.lower()
mapping = {
"urgent": "high",
"critical": "high",
"normal": "medium",
"low": "low"
}
return mapping.get(priority.lower(), "medium")

5. 为什么不用 Pydantic 或其他验证库

Pydantic 等库可以提供更强大的数据验证能力，但对于第一篇结构化输出，我们选择手动实现校验逻辑，原因是：

• 减少依赖，保持项目简洁。
• 让读者理解校验的本质，而不是依赖黑盒。
• 后续章节会引入 Pydantic，届时可以对比两种方式的优劣。

十、工程化改进：现在还不能上线

这个 Agent 能够输出结构化数据，不代表它已经是生产系统。至少还存在以下问题：

这里有一个需要提前建立的工程判断：结构化输出是 Tool Calling 的基础。

下一篇加入 Tool Calling 后，我们会看到结构化输出如何与工具调用配合，让 Agent 能够执行真实动作。

十一、本篇小结

今天完成的不是一个功能丰富的聊天页面，而是后续项目会持续复用的结构化输出能力：

提示词定义 JSON 格式
↓
模型输出结构化数据
↓
解析并校验字段
↓
格式错误时自动修复

请记住这三个结论：

十二、课后练习

请在不改变 Agent 核心结构的前提下，完成下面练习：

练习 1：增加一个任务列表功能

增加 /list 命令，打印当前会话中所有已分析的任务（保存为列表），并按优先级排序。

练习 2：实现批量任务分析

支持用户一次性输入多个任务（用换行分隔），Agent 分别分析并输出 JSON 数组。

练习 3：增加优先级冲突检测

如果用户输入的任务与已有任务的优先级冲突（如两个任务都是 high 但截止日期相同），Agent 应该提示用户并建议调整。

阶段验收

当你能够解释 SYSTEM_PROMPT、parse_json_with_retry() 和 validate_and_fix_task() 的关系，并且可以独立修改 Agent 的输出格式和校验规则时，第一阶段验收通过：

完成"智能任务助手 V1"，支持多轮对话、结构化输出和基础异常处理。

接下来，进入第二阶段：

Day 5：Tool Calling 的工作原理

✍坚持原创，求关注，点赞，收藏