优化提示词:使用多智能体系统与评估的最佳实践

本文介绍如何利用OpenAI Agents SDK和Evals构建一个多智能体提示词优化系统,自动检测并修复提示词中的矛盾、格式不明确以及少样本示例不一致等问题。系统由五个专门智能体组成,通过并行检查与条件重写流程,提升提示词的可靠性和有效性。文章详细阐述了数据模型、智能体定义、评估方法及端到端工作流,并提供了实际示例和优化建议,适合希望掌握AI工程实践的开发者学习。

技术背景

在使用大型语言模型时,提示词的质量直接影响输出效果。即使经验丰富的用户也可能无意中引入矛盾、歧义或不一致,导致模型产生不理想的结果。手动审查和优化提示词既耗时又容易出错,尤其当提示词包含复杂指令和少样本示例时。多智能体系统通过分工协作,能够系统性地识别和修复这些问题,提高提示词的可靠性和有效性。该技术适用于需要高质量、一致性输出的场景,如自动化内容生成、数据提取、对话系统等。主要限制包括依赖模型能力、需要精心设计评估集,以及可能增加计算成本。

核心原理

系统采用协作式多智能体方法,每个智能体专门负责检测或重写特定类型的问题。核心原理包括:1) 并行检测:矛盾检查器、格式检查器和少样本一致性检查器同时运行,提高效率;2) 条件重写:根据检测结果,仅在有需要时触发重写器,避免不必要的计算;3) 结构化输出:使用Pydantic模型定义输入输出格式,确保数据一致性;4) 评估驱动:通过黄金示例集和字符串检查评分器调整智能体指令和模型选择,优化性能。数据流为:输入提示词和示例消息,并行运行检查器,收集问题,若存在矛盾或格式问题则重写开发者提示,若存在少样本不一致则重写示例消息,最终输出优化后的提示词和消息。

架构分析

系统架构采用分层设计:数据模型层定义Role、ChatMessage、Issues等Pydantic模型,确保数据验证和一致性;智能体层包含五个专门Agent,每个Agent有明确的指令和输出类型;工作流层通过optimize_prompt_parallel函数编排并行检查与条件重写流程。关键设计决策包括:1) 并行化检查阶段,减少延迟;2) 条件性重写,避免不必要的模型调用;3) 使用trace记录工作流,便于调试和监控。组件关系:检查器输出Issues对象,重写器接收原始提示和问题列表,生成优化后的提示或消息。该架构支持扩展,可添加更多检查器或重写器。

完整实战教程

优化提示词

在使用AI模型时,编写有效的提示词是一项关键技能。即使是经验丰富的用户,也可能无意中引入矛盾、歧义或不一致,导致结果不理想。这里展示的系统有助于识别和修复常见问题,从而产生更可靠、更有效的提示词。

优化过程采用多智能体方法,由专门的AI智能体协作分析和重写提示词。系统自动识别并处理几类常见问题:

目标:本手册演示了如何结合使用Agents SDK和Evals来构建OpenAI提示词优化系统的早期版本。您可以使用此代码优化提示词,或者在我们的playground中使用优化器。

手册结构

本笔记本遵循以下结构:

先决条件

1. 系统概述

提示优化系统采用协作式多智能体方法来分析和改进提示。每个智能体专门负责检测或重写特定类型的问题:

  1. 开发-矛盾-检查器:扫描提示中的逻辑矛盾或不可能执行的指令,例如在同一提示中同时要求“仅使用正数”和“包含负数示例”。
  1. 格式-检查器:识别提示期望结构化输出(如 JSON、CSV 或 Markdown)但未能明确指定确切格式要求的情况。该智能体确保所有必要字段、数据类型和格式规则都被明确定义。
  1. 少样本-一致性-检查器:检查示例对话,确保助手的回复确实遵循提示中指定的规则。这可以捕获提示要求与示例展示之间的不匹配。
  1. 开发-重写器:在识别出问题后,该智能体重写提示以解决矛盾并澄清格式规范,同时保留原始意图。
  1. 少样本-重写器:更新不一致的示例回复,使其与提示中的规则对齐,确保所有示例都正确符合新的开发者提示。

通过协同工作,这些智能体可以系统地识别并修复提示中的问题。

# Import required modules
from openai import AsyncOpenAI
import asyncio
import json
import os
from enum import Enum
from typing import Any, List, Dict
from pydantic import BaseModel, Field
from agents import Agent, Runner, set_default_openai_client, trace

openai_client: AsyncOpenAI | None = None

def _get_openai_client() -> AsyncOpenAI:
    global openai_client
    if openai_client is None:
        openai_client = AsyncOpenAI(
            api_key=os.environ.get("OPENAI_API_KEY", "your-api-key"),
        )
    return openai_client

set_default_openai_client(_get_openai_client())

2. 数据模型

为了促进代理之间的结构化通信,系统使用 Pydantic 模型来定义输入和输出的预期格式。这些 Pydantic 模型有助于验证数据并确保整个工作流程的一致性。

数据模型包括:

  1. Role - 消息角色(用户/助手)的枚举
  2. ChatMessage - 表示对话中的单条消息
  3. Issues - 用于报告检测到的问题的基础模型
  4. FewShotIssues - 扩展模型,为示例消息添加重写建议
  5. MessagesOutput - 包含优化后的对话消息
  6. DevRewriteOutput - 包含改进后的开发者提示

使用 Pydantic 允许系统在流程的每一步验证所有数据是否符合预期格式。

class Role(str, Enum):
    """Role enum for chat messages."""
    user = "user"
    assistant = "assistant"

class ChatMessage(BaseModel):
    """Single chat message used in few-shot examples."""
    role: Role
    content: str

class Issues(BaseModel):
    """Structured output returned by checkers."""
    has_issues: bool
    issues: List[str]
    
    @classmethod
    def no_issues(cls) -> "Issues":
        return cls(has_issues=False, issues=[])

class FewShotIssues(Issues):
    """Output for few-shot contradiction detector including optional rewrite suggestions."""
    rewrite_suggestions: List[str] = Field(default_factory=list)
    
    @classmethod
    def no_issues(cls) -> "FewShotIssues":
        return cls(has_issues=False, issues=[], rewrite_suggestions=[])

class MessagesOutput(BaseModel):
    """Structured output returned by `rewrite_messages_agent`."""

    messages: list[ChatMessage]


class DevRewriteOutput(BaseModel):
    """Rewriter returns the cleaned-up developer prompt."""

    new_developer_message: str

3. 定义智能体

在本节中,我们使用 openai-agents 包中的 Agent 类来创建专门的 AI 智能体。查看这些智能体的定义,可以发现创建有效 AI 指令的几个最佳实践:

智能体指令中的最佳实践

  1. 明确的范围定义:每个智能体都有明确界定的目的和边界。例如,矛盾检查器只关注“真正的自相矛盾”,并明确指出“重叠或冗余不是矛盾”。
  1. 分步流程:指令提供了清晰的方法论,例如格式检查器在分析格式要求之前,首先对任务进行分类。
  1. 明确的定义:关键术语被精确定义以避免歧义。少样本一致性检查器包含一个详细的“合规性评分标准”,准确解释了合规的构成要素。
  1. 边界设定:指令指定了智能体不应执行的操作。少样本检查器明确列出了“超出范围”的内容,以防止过度标记问题。
  1. 结构化输出要求:每个智能体都有严格定义的输出格式和示例,确保优化流程的一致性。

这些原则创建了可靠、专注的智能体,它们能在优化系统中有效协作。下面我们展示了完整的智能体定义及其详细指令。

dev_contradiction_checker = Agent(
    name="contradiction_detector",
    model="gpt-4.1",
    output_type=Issues,
    instructions="""
    You are **Dev-Contradiction-Checker**.

    Goal
    Detect *genuine* self-contradictions or impossibilities **inside** the developer prompt supplied in the variable `DEVELOPER_MESSAGE`.

    Definitions
    • A contradiction = two clauses that cannot both be followed.
    • Overlaps or redundancies in the DEVELOPER_MESSAGE are *not* contradictions.

    What you MUST do
    1. Compare every imperative / prohibition against all others.
    2. List at most FIVE contradictions (each as ONE bullet).
    3. If no contradiction exists, say so.

    Output format (**strict JSON**)
    Return **only** an object that matches the `Issues` schema:

{"has_issues": <bool>,

"issues": [

"<bullet 1>",

"<bullet 2>"

]

}

Task

Decide whether the developer prompt requires a structured output (JSON/CSV/XML/Markdown table, etc.).

If so, flag any missing or unclear aspects of that format.

Steps

Categorise the task as:

  1. "conversation_only", or
  2. "structured_output_required".

For case (b):

Do NOT invent issues if unsure. be a little bit more conservative in flagging format issues

Output format

Return strictly-valid JSON following the Issues schema:

{

"has_issues": <bool>,

"issues": ["<desc 1>", "..."]

}

Maximum five issues. No extra keys or text.

""",

)

fewshot_consistency_checker = Agent(

name="fewshot_consistency_checker",

model="gpt-4.1",

output_type=FewShotIssues,

instructions="""

You are FewShot-Consistency-Checker.

Goal

Find conflicts between the DEVELOPER_MESSAGE rules and the accompanying assistant examples.

USER_EXAMPLES: <all user lines> # context only

ASSISTANT_EXAMPLES: <all assistant lines> # to be evaluated

Method

Extract key constraints from DEVELOPER_MESSAGE:

Compliance Rubric - read carefully

Evaluate only what the developer message makes explicit.

Objective constraints you must check when present:

Out-of-scope (DO NOT FLAG):

Pass/Fail rule

Empty assistant list ⇒ immediately return has_issues=false.

For each assistant example:

be a little bit more conservative in flagging few shot contradiction issues

Output format

Return JSON matching FewShotIssues:

{

"has_issues": <bool>,

"issues": ["<explanation 1>", "..."],

"rewrite_suggestions": ["<suggestion 1>", "..."] // may be []

}

List max five items for both arrays.

Provide empty arrays when none.

No markdown, no extra keys.

""",

)

dev_rewriter = Agent(

name="dev_rewriter",

model="gpt-4.1",

output_type=DevRewriteOutput,

instructions="""

You are Dev-Rewriter.

You receive:

Rewrite rules

Preserve the original intent and capabilities.

Resolve each contradiction:

If FORMAT_ISSUES is non-empty:

Do NOT change few-shot examples.

Do NOT add new policies or scope.

Output format (strict JSON)

{

"new_developer_message": "<full rewritten text>"

}

No other keys, no markdown.

""",

)

fewshot_rewriter = Agent(

name="fewshot_rewriter",

model="gpt-4.1",

output_type=MessagesOutput,

instructions="""

You are FewShot-Rewriter.

Input payload

Task

Regenerate only the assistant parts that were flagged.

User messages must remain identical.

Every regenerated assistant reply MUST comply with NEW_DEVELOPER_MESSAGE.

After regenerating each assistant reply, verify:

Output format

Return strict JSON that matches the MessagesOutput schema:

{

"messages": [

{"role": "user", "content": "..."},

{"role": "assistant", "content": "..."}

]

}

Guidelines

async def optimize_prompt_parallel(

developer_message: str,

messages: List["ChatMessage"],

) -> Dict[str, Any]:

"""

Runs contradiction, format, and few-shot checkers in parallel,

then rewrites the prompt/examples if needed.

Returns a unified dict suitable for an API or endpoint.

"""

with trace("optimize_prompt_workflow"):

1. Run all checkers in parallel (contradiction, format, fewshot if there are examples)

tasks = [

Runner.run(dev_contradiction_checker, developer_message),

Runner.run(format_checker, developer_message),

]

if messages:

fs_input = {

"DEVELOPER_MESSAGE": developer_message,

"USER_EXAMPLES": [m.content for m in messages if m.role == "user"],

"ASSISTANT_EXAMPLES": [m.content for m in messages if m.role == "assistant"],

}

tasks.append(Runner.run(fewshot_consistency_checker, json.dumps(fs_input)))

results = await asyncio.gather(*tasks)

Unpack results

cd_issues: Issues = results[0].final_output

fi_issues: Issues = results[1].final_output

fs_issues: FewShotIssues = results[2].final_output if messages else FewShotIssues.no_issues()

3. Rewrites as needed

final_prompt = developer_message

if cd_issues.has_issues or fi_issues.has_issues:

pr_input = {

"ORIGINAL_DEVELOPER_MESSAGE": developer_message,

"CONTRADICTION_ISSUES": cd_issues.model_dump(),

"FORMAT_ISSUES": fi_issues.model_dump(),

}

pr_res = await Runner.run(dev_rewriter, json.dumps(pr_input))

final_prompt = pr_res.final_output.new_developer_message

final_messages: list[ChatMessage] | list[dict[str, str]] = messages

if fs_issues.has_issues:

mr_input = {

"NEW_DEVELOPER_MESSAGE": final_prompt,

"ORIGINAL_MESSAGES": _normalize_messages(messages),

"FEW_SHOT_ISSUES": fs_issues.model_dump(),

}

mr_res = await Runner.run(fewshot_rewriter, json.dumps(mr_input))

final_messages = mr_res.final_output.messages

return {

"changes": True,

"new_developer_message": final_prompt,

"new_messages": _normalize_messages(final_messages),

"contradiction_issues": "\n".join(cd_issues.issues),

"few_shot_contradiction_issues": "\n".join(fs_issues.issues),

"format_issues": "\n".join(fi_issues.issues),

}


![Trace for the workflow](assets/images/fbb910434616dbd3.png)

### 理解优化工作流

`optimize_prompt_parallel` 函数通过并行化实现了一个最大化效率的工作流:

1. **并行问题检测**:第一阶段同时运行所有检查代理:
- `dev_contradiction_checker` 搜索提示中的逻辑矛盾
- `format_checker` 查找不明确的格式规范
- `fewshot_consistency_checker`(如果存在示例)检查提示与示例之间的不匹配

并行检查阶段之后,工作流会仔细处理依赖关系:

2. **提示重写(条件性)**:`dev_rewriter` 代理仅在检测到矛盾或格式问题时运行。该代理依赖于以下输出:
- `dev_contradiction_checker`(`cd_issues` 变量)
- `format_checker`(`fi_issues` 变量)

3. **示例重写(条件性)**:`fewshot_rewriter` 代理仅在检测到示例不一致时运行。该代理依赖于:
- 重写后的提示(必须在提示重写之后完成)
- 原始消息
- 少样本问题(`fs_issues` 变量)

## 6. 示例

让我们通过一些实际示例来看看优化系统的实际应用。

### 示例 1:修复矛盾

async def example_contradiction():

A prompt with contradictory instructions

prompt = """Quick-Start Card — Product Parser

Goal

Digest raw HTML of an e-commerce product detail page and emit concise, minified JSON describing the item.

Required fields:

name | brand | sku | price.value | price.currency | images[] | sizes[] | materials[] | care_instructions | features[]

Extraction priority:

  1. schema.org/JSON-LD blocks
  2. <meta> & microdata tags
  3. Visible DOM fallback (class hints: "product-name", "price")

Rules:

Sample skeleton (minified):

{"name":"","brand":"","sku":"","price":{"value":0,"currency":"USD"},"images":[""],"sizes":[],"materials":[],"care_instructions":"","features":[]}
Note: It is acceptable to output null for any missing field instead of an error ###"""

result = await optimize_prompt_parallel(prompt, [])

# Display the results
if result["contradiction_issues"]:
print("Contradiction issues:")
print(result["contradiction_issues"])
print()

print("Optimized prompt:")
print(result["new_developer_message"])

# Run the example
await example_contradiction()

这展示了系统如何检测并解决可能导致模型输出不一致或产生困惑的关键矛盾。

示例2:修复提示与少样本示例之间的不一致性

async def example_fewshot_fix():
    prompt = "Respond **only** with JSON using keys `city` (string) and `population` (integer)."
    
    messages = [
        {"role": "user", "content": "Largest US city?"},
        {"role": "assistant", "content": "New York City"},
        {"role": "user", "content": "Largest UK city?"},
        {"role": "assistant", "content": "{\"city\":\"London\",\"population\":9541000}"}
    ]
    
    
    print("Few-shot examples before optimization:")
    print(f"User: {messages[0]['content']}")
    print(f"Assistant: {messages[1]['content']}")
    print(f"User: {messages[2]['content']}")
    print(f"Assistant: {messages[3]['content']}")
    print()
    
    # Call the optimization API
    result = await optimize_prompt_parallel(prompt, [ChatMessage(**m) for m in messages])
    
    # Display the results
    if result["few_shot_contradiction_issues"]:
        print("Inconsistency found:", result["few_shot_contradiction_issues"])
        print()
    
    # Show the optimized few-shot examples
    optimized_messages = result["new_messages"]
    print("Few-shot examples after optimization:")
    print(f"User: {optimized_messages[0]['content']}")
    print(f"Assistant: {optimized_messages[1]['content']}")
    print(f"User: {optimized_messages[2]['content']}")
    print(f"Assistant: {optimized_messages[3]['content']}")
    
# Run the example
await example_fewshot_fix()

这一点尤为重要,因为少样本示例对模型的响应方式有着强烈的影响。如果示例不遵循所述规则,模型可能会学会忽略这些规则,转而模仿示例。通过确保提示指令与示例之间的一致性,优化系统能够创建出更可靠的提示。

示例3:在较长提示中澄清格式

async def example_format_issue():
    # A prompt with unclear or inconsistent formatting instructions
    prompt = """Task → Translate dense patent claims into 200-word lay summaries with a glossary.

Operating Steps:
1. Split the claim at semicolons, "wherein", or numbered sub-clauses.
2. For each chunk:
   a) Identify its purpose.
   b) Replace technical nouns with everyday analogies.
   c) Keep quantitative limits intact (e.g., "≥150 C").
3. Flag uncommon science terms with asterisks, and later define them.
4. Re-assemble into a flowing paragraph; do **not** broaden or narrow the claim’s scope.
5. Omit boilerplate if its removal does not alter legal meaning.

Output should follow a Markdown template:
- A summary section.
- A glossary section with the marked terms and their definitions.

Corner Cases:
- If the claim is over 5 kB, respond with CLAIM_TOO_LARGE.
- If claim text is already plain English, skip glossary and state no complex terms detected.

Remember: You are *not* providing legal advice—this is for internal comprehension only."""

    # Call the optimization API to check for format issues
    result = await optimize_prompt_parallel(prompt, [])

    # Display the results
    if result.get("format_issues"):
        print("Format issues found:", result["format_issues"])
        print()

    print("Optimized prompt:")
    print(result["new_developer_message"])

# Run the example
await example_format_issue()

这个例子展示了格式检查器如何识别并解决模糊的格式规范。提示要求输出 Markdown,而优化流程显著改进了这些格式规范。

常见问题

常见错误:1) 未设置OPENAI_API_KEY环境变量,导致客户端初始化失败;2) 消息格式不正确,如role字段拼写错误或content缺失;3) 并行任务中asyncio.gather使用不当,导致结果顺序错乱;4) 重写器输出不符合JSON schema,导致解析失败。排查方法:检查环境变量、验证输入数据、使用trace查看工作流执行情况、确保Pydantic模型与输出匹配。生产环境优化:1) 使用缓存减少重复调用;2) 调整模型选择,平衡准确性和成本;3) 增加重试机制和错误处理;4) 监控评估指标,持续优化智能体指令。

企业应用扩展

可扩展方向:1) 增加更多检查器,如语气一致性、安全性检查;2) 支持多轮对话优化;3) 集成到CI/CD流程中,自动优化提示词;4) 使用更强大的模型或微调专用模型;5) 添加用户反馈循环,持续改进优化效果。

总结

本文介绍如何利用OpenAI Agents SDK和Evals构建一个多智能体提示词优化系统,自动检测并修复提示词中的矛盾、格式不明确以及少样本示例不一致等问题。系统由五个专门智能体组成,通过并行检查与条件重写流程,提升提示词的可靠性和有效性。文章详细阐述了数据模型、智能体定义、评估方法及端到端工作流,并提供了实际示例和优化建议,适合希望掌握AI工程实践的开发者学习。

内容来源与验证

继续系统学习

站内还有更多完整实战教程

加入会员,解锁全部教程、完整代码与持续更新内容。

加入会员
查看全部提示工程教程