[Agent]OpenAI构造智能体概览[Agent] OpenAI Building Agents Overview

需要复杂判断、规则难维护或大量处理非结构化信息用agent

Agent实际是一个自动化工作流

agent组成：LLM，tools，instructions

决定LLM

LLM选择：先用最有能力的模型进行任务，然后再通过用更小的模型替代它，找到平衡点

设置评估建立模型表现baseline
看准确率
优化成本和延迟

决定tools

LLM使用API调用模型

对于没有API的app，agent操作其UI来实现操作

tools的分类：

Data：允许LLM检索需要的信息
Action：与系统进行交互
Orchestration（编排）：agent本身可以作为tool供其他agent使用

SDK：封装的代码库，是大模型调用工具，记忆和写作的基础设施

instruction的最好实践：

复用现有的文档，清晰语言，拆解任务，边缘情况处理

编排

编排有两种模式：单模型，多模型

while loop

所有编排都有一个“run”（开始），然后开始循环直到

达到退出条件（tool calls, a certain structured output, errors, or reaching a maximum number of turns.）

循环结束的情况只有：1.调用了final-output call 2。没调用任何tool并返回了结果

输入->LLM ->调用/不调用tools ->回到第二步/输出

用于复杂多agent任务的提示词工程：用一套具有延伸性的提示词可以简化维护以及增添case的成本

用多模型前： maximize a single agent’s capabilities first.

用多模型的原因：

逻辑过于复杂，单一提示词无法再扩展
工具重合度过高（overlaping）

两种多agent模式：

中心化：由一个central agent管理多个tools agent (类似调用函数)
去中心化：多个agent彼此协作(类似交接任务)

Manager_Pattern_MEDIA.png (1208×680)

Decentralized_Pattern.png (1208×680)

Guardrails机制

门禁机制,用于保护隐私，提高安全性

可以针对不同的组成部分设置对应的提示词

Guardrails类型：相关性分类，输入安检，PII（个人身份信息）筛选，有害言论审查，工具安检，基于规则的保护，输出检查

这段代码有点复杂得后面读读

from agents import (
    Agent,
    GuardrailFunctionOutput,
    InputGuardrailTripwireTriggered,
    RunContextWrapper,
    Runner,
    TResponseInputItem,
    input_guardrail,
    Guardrail,
    GuardrailTripwireTriggered,
)
from pydantic import BaseModel


class ChurnDetectionOutput(BaseModel):
    is_churn_risk: bool
    reasoning: str


churn_detection_agent = Agent(
    name="Churn Detection Agent",
    instructions=(
        "Identify if the user message indicates a potential customer churn risk."
    ),
    output_type=ChurnDetectionOutput,
)


@input_guardrail
async def churn_detection_tripwire(
    ctx: RunContextWrapper[None],
    agent: Agent,
    input: str | list[TResponseInputItem],
) -> GuardrailFunctionOutput:
    result = await Runner.run(
        churn_detection_agent,
        input,
        context=ctx.context,
    )

    return GuardrailFunctionOutput(
        output_info=result.final_output,
        tripwire_triggered=result.final_output.is_churn_risk,
    )


customer_support_agent = Agent(
    name="Customer Support Agent",
    instructions=(
        "You are a customer support agent. You help customers with their questions."
    ),
    input_guardrails=[
        Guardrail(guardrail_function=churn_detection_tripwire),
    ],
)


async def main():
    # This should be ok
    await Runner.run(customer_support_agent, "Hello!")
    print("Hello message passed")

    # This should trip the guardrail
    try:
        await Runner.run(
            customer_support_agent,
            "I think I might cancel my subscription",
        )
        print("Guardrail didn't trip - this is unexpected")
    except GuardrailTripwireTriggered:
        print("Churn detection guardrail tripped")

人工干预

兜底
记录每一次人工干预可以以此更新迭代

触发人工介入：

超出失败阈值（Exceeding failure thresholds）： 这与大模型底层的 while 循环直接相关。必须在系统中设定明确的重试上限。
高风险操作（High-risk actions）： 对于具有破坏性、不可逆转或涉及资金的功能，AI 只能做“提案（Proposal）”，不能做“决断（Execution）”，这类需要人工批准

Use an agent when complex judgment is required, rules are hard to maintain, or there is a large amount of unstructured information to process.

An agent is essentially an automated workflow.

Agent composition: LLM, tools, instructions

Deciding on the LLM

LLM selection: start by using the most capable model for the task, then find the balance point by replacing it with smaller models.

Set up evaluations to establish a baseline for model performance
Look at accuracy
Optimize cost and latency

Deciding on tools

LLMs use APIs to call models.

For apps without APIs, the agent operates their UI to perform actions.

Categories of tools:

Data: allows the LLM to retrieve needed information
Action: interacts with systems
Orchestration: the agent itself can serve as a tool for other agents to use

SDK: an encapsulated code library, the infrastructure for large model tool calling, memory, and writing

Best practices for instructions:

Reuse existing documentation, use clear language, break down tasks, handle edge cases

Orchestration

There are two orchestration modes: single-model and multi-model

while loop

All orchestration has a "run" (start), then begins a loop until

an exit condition is reached (tool calls, a certain structured output, errors, or reaching a maximum number of turns.)

The loop ends only in these cases: 1. a final-output call is invoked 2. no tool is called and a result is returned

Input -> LLM -> call / do not call tools -> return to step two / output

Prompt engineering for complex multi-agent tasks: using an extensible set of prompts can simplify maintenance and reduce the cost of adding cases

Before using multiple models: maximize a single agent’s capabilities first.

Reasons to use multiple models:

The logic is too complex and a single prompt can no longer scale
Tool overlap is too high (overlapping)

Two multi-agent modes:

Centralized: a central agent manages multiple tool agents (similar to calling functions)
Decentralized: multiple agents collaborate with one another (similar to handing off tasks)

Manager_Pattern_MEDIA.png (1208×680)

Decentralized_Pattern.png (1208×680)

Guardrails mechanism

Access-control mechanism, used to protect privacy and improve security

Corresponding prompts can be set for different components

Types of guardrails: relevance classification, input screening, PII (personally identifiable information) filtering, harmful speech moderation, tool screening, rule-based protection, output checking

This code is a bit complex and should be read later

from agents import (
    Agent,
    GuardrailFunctionOutput,
    InputGuardrailTripwireTriggered,
    RunContextWrapper,
    Runner,
    TResponseInputItem,
    input_guardrail,
    Guardrail,
    GuardrailTripwireTriggered,
)
from pydantic import BaseModel


class ChurnDetectionOutput(BaseModel):
    is_churn_risk: bool
    reasoning: str


churn_detection_agent = Agent(
    name="Churn Detection Agent",
    instructions=(
        "Identify if the user message indicates a potential customer churn risk."
    ),
    output_type=ChurnDetectionOutput,
)


@input_guardrail
async def churn_detection_tripwire(
    ctx: RunContextWrapper[None],
    agent: Agent,
    input: str | list[TResponseInputItem],
) -> GuardrailFunctionOutput:
    result = await Runner.run(
        churn_detection_agent,
        input,
        context=ctx.context,
    )

    return GuardrailFunctionOutput(
        output_info=result.final_output,
        tripwire_triggered=result.final_output.is_churn_risk,
    )


customer_support_agent = Agent(
    name="Customer Support Agent",
    instructions=(
        "You are a customer support agent. You help customers with their questions."
    ),
    input_guardrails=[
        Guardrail(guardrail_function=churn_detection_tripwire),
    ],
)


async def main():
    # This should be ok
    await Runner.run(customer_support_agent, "Hello!")
    print("Hello message passed")

    # This should trip the guardrail
    try:
        await Runner.run(
            customer_support_agent,
            "I think I might cancel my subscription",
        )
        print("Guardrail didn't trip - this is unexpected")
    except GuardrailTripwireTriggered:
        print("Churn detection guardrail tripped")

Human intervention

Fallback
Record every human intervention so it can be used to update and iterate

Triggers for human intervention:

Exceeding failure thresholds: This is directly related to the underlying while loop of the large model. Clear retry limits must be set in the system.
High-risk actions: For destructive, irreversible, or fund-related functions, AI can only make a "Proposal" and cannot make an "Execution." These require human approval.