到目前为止，我们的Agent的输出都是自由文本。LLM给出一个回答，我们把这段文本直接展示给用户。这在对话场景中没问题，但在很多实际应用中，你需要的是机器可处理的数据。
考虑以下场景：
场景一：信息提取
用户说："帮我从这段客户反馈中提取关键信息。"
你期望的输出不是一段自然语言总结，而是一个结构化的对象：
{
"sentiment": "negative",
"key_issues": ["配送延迟", "包装破损"],
"product": "无线耳机",
"customer_priority": "high"
}
场景二：Agent间的数据传递
在多Agent系统中，Agent A需要把结果传递给Agent B。如果Agent A的输出是自由文本，Agent B需要额外一步"理解"来提取关键信息。如果Agent A的输出是结构化的，Agent B可以直接使用。
场景三：与外部系统的集成
Agent需要把结果写入数据库、调用API、生成报告。这些操作都需要确定的数据格式，而非自然语言。
核心问题：如何让LLM输出结构化的数据，而不仅仅是"看起来像结构化"的文本？
实现结构化输出有三种主要方式：
方式一：Prompt引导
在系统提示词中要求LLM输出JSON格式：
你的所有回答必须以JSON格式输出，格式如下：
然后用正则表达式或JSON解析器提取。
优点：实现简单，任何模型都支持。 缺点：不可靠。LLM有时会忽略格式要求，输出前后带多余文本（"好的，以下是JSON：..."），或者JSON格式错误（尾逗号、引号不匹配）。
适用场景：原型验证、对格式要求不严格的场景。
方式二：Function Calling的变体 让LLM调用一个"输出结果"的函数，参数就是结构化数据。
优点：利用Function Calling的天然结构化能力，比纯Prompt引导更可靠。 缺点：需要在工具列表中增加一个"伪工具"，语义上不够清晰。
适用场景：模型中已有工具调用能力，但还没有原生的结构化输出支持。
方式三：原生结构化输出（推荐） OpenAI在2024年推出了JSON Schema约束功能：你给LLM一个JSON Schema，它保证输出符合这个Schema。这是通过API参数response_format实现的。
优点：最可靠，由API层面保证格式正确。 缺点：目前不是所有Provider都支持；对模型能力有一定要求。
适用场景：生产环境、需要高可靠性的场景。
方式三的深入实现 让我们在Harness中实现原生结构化输出。
第一步：定义输出Schema
from typing import Dict, Any, Optional, List
class StructuredOutputConfig:
"""结构化输出的配置"""
def __init__(self, json_schema: Dict[str, Any],
strict: bool = True):
self.json_schema = json_schema
self.strict = strict
一个示例Schema：
CUSTOMER_FEEDBACK_SCHEMA = {
"type": "object",
"properties": {
"sentiment": {
"type": "string",
"enum": ["positive", "neutral", "negative"],
"description": "客户情感倾向"
},
"key_issues": {
"type": "array",
"items": {"type": "string"},
"description": "关键问题列表"
},
"product": {
"type": "string",
"description": "涉及的产品名称"
},
"customer_priority": {
"type": "string",
"enum": ["low", "medium", "high"],
"description": "客户优先级"
},
"requires_followup": {
"type": "boolean",
"description": "是否需要跟进"
}
},
"required": ["sentiment", "key_issues", "requires_followup"],
"additionalProperties": False
}
第二步：在Runtime中集成 修改AgentRuntime，添加结构化输出支持：
class AgentRuntime:
def __init__(self, config, tool_registry, memory,
output_schema: Optional[Dict[str, Any]] = None):
self.config = config
self.tools = tool_registry
self.memory = memory
self.output_schema = output_schema
self.client = OpenAI(
base_url=config.llm.base_url,
api_key=config.llm.api_key,
)
def run(self, user_input: str) -> AgentResult:
start_time = time.time()
self.memory.initialize(user_input)
for iteration in range(self.config.runtime.max_iterations):
elapsed = time.time() - start_time
if elapsed > self.config.runtime.timeout_seconds:
return AgentResult(
status="timeout",
iterations=iteration,
duration_ms=int(elapsed * 1000),
termination_reason="超时"
)
# 构建API调用参数
api_params = {
"model": self.config.llm.model,
"messages": self.memory.get_context(),
"temperature": self.config.llm.temperature,
"max_tokens": self.config.llm.max_tokens,
}
# 如果有工具，添加工具定义
tool_schemas = self.tools.get_schemas()
if tool_schemas:
api_params["tools"] = tool_schemas
# 如果配置了结构化输出，添加响应格式约束
if self.output_schema:
api_params["response_format"] = {
"type": "json_schema",
"json_schema": {
"name": "agent_output",
"schema": self.output_schema,
"strict": True,
}
}
response = self.client.chat.completions.create(**api_params)
message = response.choices[0].message
if message.tool_calls:
# 处理工具调用（与之前相同）
self.memory.add_llm_message(message)
calls = self._parse_tool_calls(message)
feedbacks = self._execute_tool_calls(calls)
for feedback in feedbacks:
self.memory.messages.append(feedback)
else:
# 正常终止
content = message.content or ""
# 如果配置了结构化输出，解析JSON
if self.output_schema:
try:
parsed = json.loads(content)
return AgentResult(
status="success",
content=content,
structured_output=parsed,
iterations=iteration + 1,
duration_ms=int((time.time() - start_time) * 1000),
termination_reason="正常终止（结构化输出）"
)
except json.JSONDecodeError as e:
# 理论上不应该发生（API层面保证），但做好兜底
return AgentResult(
status="error",
content=f"结构化输出解析失败：{e}",
iterations=iteration + 1,
duration_ms=int((time.time() - start_time) * 1000),
termination_reason="结构化输出解析异常"
)
else:
return AgentResult(
status="success",
content=content,
iterations=iteration + 1,
duration_ms=int((time.time() - start_time) * 1000),
termination_reason="正常终止"
)
return AgentResult(
status="max_iterations",
content="已达到最大循环次数",
iterations=self.config.runtime.max_iterations,
duration_ms=int((time.time() - start_time) * 1000),
termination_reason="达到最大迭代次数"
)
同时更新AgentResult：
class AgentResult:
def __init__(self, status: str, content: str = "",
structured_output: Optional[Dict[str, Any]] = None,
iterations: int = 0, duration_ms: int = 0,
termination_reason: str = ""):
self.status = status
self.content = content
self.structured_output = structured_output # 新增：结构化的输出
self.iterations = iterations
self.duration_ms = duration_ms
self.termination_reason = termination_reason
def to_dict(self):
return {
"status": self.status,
"content": self.content,
"structured_output": self.structured_output,
"iterations": self.iterations,
"duration_ms": self.duration_ms,
"termination_reason": self.termination_reason,
}
第三步：使用示例
# examples/structured_agent.py
# 定义输出Schema
FEEDBACK_SCHEMA = {
"type": "object",
"properties": {
"sentiment": {
"type": "string",
"enum": ["positive", "neutral", "negative"],
"description": "客户情感倾向"
},
"key_issues": {
"type": "array",
"items": {"type": "string"},
"description": "关键问题列表"
},
"product": {
"type": "string",
"description": "涉及的产品名称"
},
"customer_priority": {
"type": "string",
"enum": ["low", "medium", "high"],
"description": "客户优先级"
},
"requires_followup": {
"type": "boolean",
"description": "是否需要跟进"
}
},
"required": ["sentiment", "key_issues", "requires_followup"],
"additionalProperties": False
}
# 创建带结构化输出的Runtime
runtime = AgentRuntime(
config=config,
tool_registry=registry,
memory=memory,
output_schema=FEEDBACK_SCHEMA, # 传入Schema
)
# 运行
result = runtime.run(
"客户反馈：我买的无线耳机三天才到货，而且包装破了。很失望。"
)
# 使用结构化输出
if result.structured_output:
print(f"情感：{result.structured_output['sentiment']}")
print(f"关键问题：{result.structured_output['key_issues']}")
print(f"需要跟进：{result.structured_output['requires_followup']}")
输出：
情感：negative
关键问题：['配送延迟', '包装破损']
需要跟进：true
这是一个微妙的问题。答案是：取决于Provider的实现。
在OpenAI的API中，当你在请求中同时指定了tools和response_format时：
如果LLM决定调用工具，它输出的是工具调用请求（而非符合Schema的内容）
如果LLM决定直接回答，它输出的是符合Schema的内容
这意味着：结构化输出约束的是"最终回答"的格式，而非工具调用的格式。 工具调用的格式始终是Function Calling的标准格式。
这对我们的Harness来说正好——它允许Agent在循环中自由使用工具，最终的回答则被约束在预定义的Schema中。
如果你的Provider不支持原生的response_format，可以用方式二（Function Calling变体）：
# 定义一个"提交结果"的工具
submit_result_schema = {
"type": "object",
"properties": {
"sentiment": {"type": "string", "enum": ["positive", "neutral", "negative"]},
"key_issues": {"type": "array", "items": {"type": "string"}},
"requires_followup": {"type": "boolean"},
},
"required": ["sentiment", "key_issues", "requires_followup"],
}
def submit_result(sentiment: str, key_issues: list, requires_followup: bool) -> str:
"""提交最终的分析结果"""
# 这个函数的"返回"不重要，重要的是参数
return "结果已提交"
registry.register(
name="submit_result",
description="提交最终的分析结果。在你完成分析后，必须调用此工具提交结果。",
func=submit_result,
parameters=submit_result_schema,
)
然后在系统提示词中明确要求：
你是一个客户反馈分析助手。你的任务是分析客户反馈并提交结构化的结果。
分析完成后，你必须调用 submit_result 工具来提交你的分析结果。
不要直接在对话中输出分析结果，必须通过 submit_result 工具提交。
当LLM调用submit_result时，它的参数就是结构化的结果。Runtime可以在执行submit_result时捕获参数，将其作为结构化输出。
def _execute_tool_calls(self, calls):
results = []
for call in calls:
if call.name == "submit_result":
# 捕获结构化输出
self.structured_output = call.arguments
# 不执行实际的工具函数，直接返回确认
results.append({
"tool_call_id": call.call_id,
"role": "tool",
"content": "结果已提交"
})
continue
# ... 其他工具的正常执行 ...
return results
这种方式的优点：兼容性最好，任何支持Function Calling的模型都能用。 缺点：需要在提示词中"哄"LLM调用提交工具，有时它会忘记。
有时候你无法使用方式二或方式三（比如模型不支持Function Calling），只能从自由文本中提取结构化数据。这时需要防御性的解析：
import re
import json
def extract_json_from_text(text: str) -> Optional[Dict[str, Any]]:
"""从自由文本中提取JSON对象"""
# 尝试直接解析
try:
return json.loads(text)
except json.JSONDecodeError:
pass
# 尝试提取```json ... ```代码块
json_block = re.search(r'```json\s*\n(.*?)\n```', text, re.DOTALL)
if json_block:
try:
return json.loads(json_block.group(1))
except json.JSONDecodeError:
pass
# 尝试提取{...}（贪婪匹配最外层）
brace_match = re.search(r'\{.*\}', text, re.DOTALL)
if brace_match:
try:
return json.loads(brace_match.group(0))
except json.JSONDecodeError:
pass
# 尝试提取[...]（数组）
bracket_match = re.search(r'\\[.*\\]', text, re.DOTALL)
if bracket_match:
try:
return json.loads(bracket_match.group(0))
except json.JSONDecodeError:
pass
return None
这个函数体现了"防御性编程"的思想：对LLM的输出做多层降级处理。每一层都尝试从文本中提取结构化数据，如果失败就尝试下一种模式。最终如果全部失败，返回None让调用方决定如何处理。
| 场景 | 推荐方式 | 原因 |
|---|---|---|
Provider支持response_format | 方式三（原生） | 最可靠，API层面保证 |
| Provider不支持，但有Function Calling | 方式二（工具变体） | 兼容性好，可靠性可接受 |
| 模型没有Function Calling能力 | 方式一（Prompt引导+解析） | 唯一选择，但要做好失败处理 |
| 简单原型，格式要求不严格 | 方式一 | 实现最快 |
结合本节和前面各节的内容，构建一个完整的信息提取Agent：
# examples/extraction_agent.py
import os
import json
from harness.config import HarnessConfig, LLMConfig, RuntimeConfig
from harness.tools import ToolRegistry
from harness.memory import MemorySystem
from harness.runtime import AgentRuntime
# 配置
config = HarnessConfig(
llm=LLMConfig(
api_key=os.getenv("OPENAI_API_KEY", ""),
model="gpt-4o-mini",
temperature=0.0,
),
runtime=RuntimeConfig(
max_iterations=5,
timeout_seconds=30,
retry_attempts=1,
)
)
# 输出Schema
EXTRACTION_SCHEMA = {
"type": "object",
"properties": {
"entities": {
"type": "array",
"items": {
"type": "object",
"properties": {
"name": {"type": "string", "description": "实体名称"},
"type": {"type": "string", "enum": ["person", "organization", "location", "product", "other"], "description": "实体类型"},
"mentions": {"type": "integer", "description": "提及次数"}
},
"required": ["name", "type", "mentions"],
"additionalProperties": False
},
"description": "文本中的实体列表"
},
"summary": {"type": "string", "description": "文本摘要，不超过50字"},
"language": {"type": "string", "description": "文本语言"}
},
"required": ["entities", "summary", "language"],
"additionalProperties": False
}
# 工具（提取场景可能不需要工具，但保留空注册表以保持结构一致）
registry = ToolRegistry()
# 记忆
memory = MemorySystem(
system_prompt=(
"你是一个信息提取助手。分析用户提供的文本，"
"提取其中的实体信息并生成摘要。"
)
)
# Runtime（带结构化输出）
runtime = AgentRuntime(
config=config,
tool_registry=registry,
memory=memory,
output_schema=EXTRACTION_SCHEMA,
)
# 运行
text = """
苹果公司今天宣布将在北京开设新的研发中心。
CEO蒂姆·库克表示，中国市场对苹果来说非常重要。
这是苹果在北京的第三个研发中心。
"""
result = runtime.run(text)
if result.structured_output:
print(json.dumps(result.structured_output, ensure_ascii=False, indent=2))
输出示例：
{
"entities": [
{
"name": "苹果公司",
"type": "organization",
"mentions": 3
},
{
"name": "北京",
"type": "location",
"mentions": 2
},
{
"name": "蒂姆·库克",
"type": "person",
"mentions": 1
},
{
"name": "中国市场",
"type": "other",
"mentions": 1
}
],
"summary": "苹果公司在北京设立第三个研发中心，CEO库克强调中国市场的重要性。",
"language": "中文"
}
拿到了结构化输出后，还应该做一件事：验证输出是否符合预期。虽然API层面保证了格式符合Schema，但语义层面的验证仍然需要：
def validate_structured_output(output: Dict[str, Any]) -> List[str]:
"""验证结构化输出的语义正确性"""
warnings = []
# 示例：检查实体列表是否为空
if "entities" in output and len(output["entities"]) == 0:
warnings.append("未提取到任何实体，请检查输入文本是否包含可识别的实体")
# 示例：检查摘要长度
if "summary" in output and len(output["summary"]) > 50:
warnings.append(f"摘要长度（{len(output['summary'])}字）超过50字限制")
return warnings
结构化输出让Agent从"会说话"升级为"能产出可用的数据"。三种实现方式各有适用场景：
Prompt引导最简单，但最不可靠，适合原型
Function Calling变体兼容性好，需要"哄"LLM调用提交工具
原生response_format最可靠，是生产环境的首选
无论使用哪种方式，都需要防御性的解析和验证——LLM的输出永远不应该被完全信任。
在下一节中，我们将讨论日志与调试基础——如何让Agent的执行过程可见、可追踪、可诊断。这是Harness可观测性的起点。
评论专区
评论加载中...登录后即可发表评论