Agents-SDK智能体开发及部署

OpenAI Agents-SDK 从入门到企业级实践教学笔记

本文面向AI智能体应用开发者，系统讲解OpenAI官方企业级Multi-Agent开发框架Agents-SDK的核心知识点。内容从环境搭建起步，逐步深入至单智能体、多智能体开发，最终覆盖企业级项目构建与部署，所有代码均采用官方推荐的异步编程范式与最佳实践。

---

一、Agents-SDK 核心介绍

1.1 什么是Agents-SDK

Agents-SDK是OpenAI于2024年底正式推出的企业级多智能体开发框架，作为Swarm的替代方案，提供构建、编排和部署AI智能体系统的完整解决方案。它深度集成OpenAI模型能力，支持智能体自主决策、工具调用、跨智能体任务移交（Handoff）以及全链路追踪，专为生产环境设计。

1.2 核心优势

原生OpenAI集成：无缝对接OpenAI全系列模型（GPT-4o, o1等），享受官方最新特性支持企业级可靠性：内置安全校验、权限控制和操作审计，满足生产环境SLA要求智能体编排：支持确定性（规则驱动）和动态式（LLM自主决策）两种任务分发模式简洁的工具系统：通过 @function_tool装饰器快速集成Python函数，自动处理参数解析自动记忆管理：内置 Memory系统，自动维护对话上下文，无需手动管理历史列表生产级追踪：集成OpenAI Traces，完整记录智能体决策链路与工具调用过程

1.3 适用场景

客户服务系统（多技能客服智能体协作）、企业数据分析平台（数据处理-分析-可视化流水线）、智能办公助理（日程-邮件-文档协同）、研发辅助工具（需求-编码-测试智能体链）等复杂企业级应用。

---

二、环境准备与安装步骤

2.1 前置依赖

Python 3.9+：SDK基于现代Python异步特性构建OpenAI API密钥：需获取OpenAI API Key（推荐使用环境变量管理）虚拟环境：强烈推荐使用 venv或 conda隔离项目依赖

2.2 详细安装步骤

# 步骤1：创建并激活虚拟环境
python -m venv agents-env

# Windows PowerShell
.agents-envScriptsActivate.ps1
# macOS/Linux
source agents-env/bin/activate

# 步骤2：安装核心依赖（仅需两个包）
pip install openai-agents  # 主框架
pip install python-dotenv  # 环境变量管理（可选但推荐）

# 步骤3：配置API密钥
# 创建.env文件（切勿硬编码到代码中）
echo "OPENAI_API_KEY=sk-你的密钥" > .env

# 步骤4：验证环境
# 创建test.py并运行

验证脚本（test.py）：

import asyncio
import os
from dotenv import load_dotenv
from openai import AsyncOpenAI
from agents import Agent, Runner

load_dotenv()

async def test_environment():
    client = AsyncOpenAI(api_key=os.getenv("OPENAI_API_KEY"))
    agent = Agent(name="test", model="gpt-4o-mini", instructions="回复'环境正常'")
    result = await Runner.run(agent, "测试")
    print(f"✅ 环境配置成功！模型响应：{result.final_output}")

if __name__ == "__main__":
    asyncio.run(test_environment())

---

三、Agents-SDK 基本使用

3.1 核心组件与参数

Agents-SDK三大核心组件： Agent（智能体实体）、 Runner（异步执行器）、 function_tool（工具装饰器）。

3.1.1 Agent核心参数

Agent(
    name="智能体名称",           # 用于日志追踪与标识
    model="gpt-4o",              # 模型ID字符串（直接指定，无需包装类）
    instructions="系统提示词",    # 角色定义与行为准则
    tools=[tool1, tool2],        # 工具函数列表（注意是列表不是元组）
    handoffs=[agent1, agent2],   # 可移交的目标智能体列表
    model_settings=ModelSettings( # 模型细粒度配置（可选）
        temperature=0.2,
        max_tokens=2000
    )
)

3.1.2 Runner.run核心参数

result = await Runner.run(
    agent=agent_instance,        # Agent实例
    input="用户输入",             # 字符串输入
    memory=memory_instance       # Memory对象（自动管理，无需手动维护）
)

返回值： RunResult对象包含：

final_output: str - 智能体最终文本响应 new_memory: Memory - 更新后的记忆对象（自动包含本轮对话） trace: Trace - 完整执行追踪信息（用于调试）

3.2 基础多轮对话实现

官方推荐范式：使用 Memory对象而非手动维护列表，利用 Runner.run返回的 new_memory自动更新上下文。

import asyncio
import os
from dotenv import load_dotenv
from openai import AsyncOpenAI
from agents import Agent, Runner

load_dotenv()

async def main():
    # 1. 初始化OpenAI异步客户端
    client = AsyncOpenAI(api_key=os.getenv("OPENAI_API_KEY"))
    
    # 2. 定义基础智能体（直接指定模型字符串）
    agent = Agent(
        name="BasicAssistant",
        model="gpt-4o-mini",
        instructions="你是友好助手，回答简洁准确。无法解答时明确说明，不编造。"
    )
    
    # 3. 初始化记忆（自动管理）
    memory = []
    
    print("智能助手：您好！（输入'退出'结束）")
    
    while True:
        user_input = input("
用户：")
        if user_input.strip() == "退出":
            break
        
        # 4. 运行智能体（传入当前记忆）
        result = await Runner.run(agent, user_input, memory=memory)
        
        # 5. 输出响应
        print(f"助手：{result.final_output}")
        
        # 6. 自动更新记忆（关键：使用返回的new_memory）
        memory = result.new_memory

if __name__ == "__main__":
    asyncio.run(main())

---

四、单智能体创建与开发

4.1 无外部工具的智能体

适用于纯对话场景，核心是利用 instructions精准定义角色与边界。

# agents/ecommerce_agent.py
from agents import Agent

def create_ecommerce_agent() -> Agent:
    return Agent(
        name="EcommerceCustomerServiceAgent",
        model="gpt-4o",
        instructions="""你是电商平台客服助手，职责如下：
1. 解答订单查询、退换货政策、物流状态等基础问题
2. 用户提及订单号时，主动确认订单商品名称与下单时间
3. 使用亲切口语化表达，禁用专业术语
4. 无法解答的问题（如支付异常），引导用户联系人工客服：400-123-4567
5. 每轮回应不超过3句话""",
        model_settings={"temperature": 0.2}  # 精准控制
    )

使用示例：

# main.py
from agents.ecommerce_agent import create_ecommerce_agent

async def ecommerce_chat():
    agent = create_ecommerce_agent()
    memory = []
    
    print("客服助手：您好！我是您的专属客服")
    
    while True:
        user = input("用户：")
        if user in ["退出", "再见"]:
            break
        
        result = await Runner.run(agent, user, memory=memory)
        print(f"客服：{result.final_output}")
        memory = result.new_memory

4.2 集成外部工具的智能体

通过 @function_tool装饰器将Python函数转为智能体可调用的工具。关键规范：必须提供清晰的docstring，包含参数说明与返回值描述。

# tools/logistics.py
from agents import function_tool

@function_tool
def query_logistics(order_id: str) -> str:
    """
    查询电商订单物流状态
    
    参数:
        order_id: 9位数字订单号（纯数字字符串）
        
    返回:
        包含物流节点与预计送达时间的描述文本
    """
    # 模拟数据库（生产环境应调用真实API）
    logistics_db = {
        "123456789": "【北京朝阳区】已揽收，预计2025-10-20送达",
        "987654321": "【上海浦东】派送中，快递员：张师傅 138****1234"
    }
    
    return logistics_db.get(
        order_id, 
        f"未查询到订单{order_id}，请核对后重试"
    )

# agents/logistics_agent.py
from agents import Agent
from tools.logistics import query_logistics

def create_logistics_agent() -> Agent:
    return Agent(
        name="LogisticsAgent",
        model="gpt-4o",
        instructions="""你是物流查询专家，遵循以下流程：
1. 用户询问物流时，若未提供订单号，主动询问
2. 获取订单号后，**立即调用**query_logistics工具，无需确认
3. 将工具返回的原始信息转换为自然语言回复
4. 若查询失败，提示核对订单号或联系人工客服""",
        tools=[query_logistics],  # 工具列表
        model_settings={"temperature": 0.1}
    )

关键特性：智能体自动决定是否调用工具、提取参数、处理返回值，开发者无需编写调用逻辑。

---

五、多智能体系统与任务移交

5.1 Handoff机制原理

Handoff是Agents-SDK的核心创新，当智能体判断任务超出能力范围时，自动将对话上下文与控制权移交给目标智能体。记忆（Memory）全程自动传递，确保上下文不丢失。

# agents/return_agent.py
from agents import Agent

def create_return_agent() -> Agent:
    return Agent(
        name="ReturnRefundAgent",
        model="gpt-4o",
        instructions="""你是退换货处理专家，职责：
1. 解答退换货政策、流程、条件
2. 用户提及订单时，确认订单号与退换原因
3. 引导用户完成申请：登录APP→我的订单→申请退换货
4. 明确告知需准备的材料：完整包装、发票、赠品"""
    )

# agents/triage_agent.py
from agents import Agent

def create_triage_agent(model="gpt-4o") -> Agent:
    logistics_agent = create_logistics_agent()
    return_agent = create_return_agent()
    
    return Agent(
        name="TriageAgent",
        model=model,
        instructions="""你是电商服务分诊员，遵循：
1. **精准识别**用户核心需求（物流查询/退换货/其他）
2. **物流需求** → 移交LogisticsAgent
3. **退换货需求** → 移交ReturnRefundAgent
4. **需求不明** → 主动追问一次，再移交
5. 移交时**禁止**额外说明，直接转移控制权""",
        handoffs=[logistics_agent, return_agent]  # 可移交目标列表
    )

5.2 完整多智能体系统

# main.py
from agents.triage_agent import create_triage_agent

async def multi_agent_system():
    triage = create_triage_agent()
    memory = []
    
    print("🤖 综合客服：您好！请问有什么可以帮助您？")
    
    while True:
        user = input("
用户：")
        if user in ["退出", "结束"]:
            break
        
        # 始终从分诊智能体进入
        result = await Runner.run(triage, user, memory=memory)
        
        print(f"助手：{result.final_output}")
        memory = result.new_memory

运行效果示例：

用户：我想查订单123456789的物流
→ TriageAgent识别为物流需求 → 自动移交LogisticsAgent → 调用工具 → 返回物流信息

用户：这个商品能退吗？
→ TriageAgent识别为退换货 → 移交ReturnRefundAgent → 专业解答

---

六、企业级项目架构

6.1 推荐项目结构

enterprise-agent/
├── config/
│   ├── __init__.py
│   └── settings.py          # 环境配置与模型初始化
├── agents/
│   ├── __init__.py
│   ├── triage.py            # 分诊智能体
│   ├── scheduler.py         # 日程管理智能体
│   └── info.py              # 企业信息查询智能体
├── tools/
│   ├── __init__.py
│   ├── calendar.py          # 日程工具（对接企业日历API）
│   └── knowledge.py         # 知识库查询工具
├── utils/
│   ├── __init__.py
│   ├── logger.py            # 结构化日志
│   └── memory_store.py      # 记忆持久化（Redis/DB）
└── main.py                  # 应用入口

6.2 核心模块实现

# config/settings.py
import os
from openai import AsyncOpenAI
from agents import Agent, ModelSettings

load_dotenv()

def get_client() -> AsyncOpenAI:
    """获取OpenAI异步客户端"""
    return AsyncOpenAI(
        api_key=os.getenv("OPENAI_API_KEY"),
        # 如需使用第三方兼容API，可添加 base_url="https://api.deepseek.com/v1"
    )

def get_model_settings() -> ModelSettings:
    """企业级模型配置"""
    return ModelSettings(
        temperature=0.2,
        max_tokens=2000,
        top_p=0.9
    )

# tools/calendar.py
from agents import function_tool
from pydantic import BaseModel

class ScheduleParams(BaseModel):
    """参数校验（推荐做法）"""
    title: str
    start_time: str  # ISO format
    end_time: str
    participants: list[str] = []

@function_tool
def create_schedule(params: ScheduleParams) -> str:
    """
    创建企业日程（使用Pydantic校验参数）
    """
    # 实际对接企业日历API（如Microsoft Graph, Google Calendar）
    return f"✅ 日程【{params.title}】创建成功
时间：{params.start_time}至{params.end_time}"

# agents/scheduler.py
from agents import Agent
from tools.calendar import create_schedule
from config.settings import get_model_settings

def create_scheduler_agent() -> Agent:
    return Agent(
        name="EnterpriseScheduler",
        model="gpt-4o",
        instructions="""你是企业日程管理专家，遵循：
1. 创建日程必须获取：标题、开始时间、结束时间
2. 时间格式：YYYY-MM-DDTHH:MM:SS（如2025-10-18T14:30:00）
3. 缺失信息时，**逐一**询问，每次只问一个
4. 获取完整信息后，调用create_schedule工具""",
        tools=[create_schedule],
        model_settings=get_model_settings()
    )

# main.py（生产级入口）
import asyncio
import os
from dotenv import load_dotenv
from agents import Runner
from config.settings import get_client
from agents.triage import create_triage_agent
from utils.logger import setup_logging
from utils.memory_store import RedisMemoryStore

load_dotenv()

async def main():
    # 初始化
    logger = setup_logging()
    client = get_client()
    
    # 记忆持久化（Redis）
    memory_store = RedisMemoryStore(
        url=os.getenv("REDIS_URL", "redis://localhost:6379")
    )
    
    # 加载或创建记忆
    session_id = os.getenv("SESSION_ID", "default")
    memory = await memory_store.load(session_id) or []
    
    # 创建智能体
    triage_agent = create_triage_agent()
    
    logger.info(f"企业智能助理启动，会话ID: {session_id}")
    
    try:
        while True:
            user = input("
员工：")
            if user in ["退出", "下班"]:
                break
            
            # 执行
            result = await Runner.run(
                triage_agent, 
                user, 
                memory=memory
            )
            
            print(f"助理：{result.final_output}")
            
            # 更新并持久化记忆
            memory = result.new_memory
            await memory_store.save(session_id, memory)
            
    except Exception as e:
        logger.error(f"会话异常: {e}", exc_info=True)
        print("⚠️ 服务暂时异常，请稍后重试")

if __name__ == "__main__":
    asyncio.run(main())

---

七、生产环境部署方案

7.1 Docker容器化（推荐）

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

# 安装系统依赖
RUN apt-get update && apt-get install -y 
    gcc 
    && rm -rf /var/lib/apt/lists/*

# 复制依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 复制代码
COPY . .

# 非root用户运行（安全最佳实践）
RUN useradd -m -u 1000 agentuser
USER agentuser

CMD ["python", "main.py"]

requirements.txt：

openai-agents>=0.5.0
openai>=1.55.0
python-dotenv>=1.0.0
redis>=5.0.0
pydantic>=2.0.0

构建与运行：

docker build -t enterprise-agent:v1 .
docker run -d 
  --env-file .env 
  -v $(pwd)/logs:/app/logs 
  --restart unless-stopped 
  enterprise-agent:v1

7.2 AWS生产部署

架构建议：

ECS Fargate：无服务器容器编排，自动扩缩容ElastiCache Redis：记忆持久化存储Secrets Manager：安全存储API密钥CloudWatch：日志与监控ALB：负载均衡与SSL终止

部署步骤：

推送镜像到ECR

aws ecr get-login-password | docker login --username AWS --password-stdin <account>.dkr.ecr.<region>.amazonaws.com
docker tag enterprise-agent:v1 <account>.dkr.ecr.<region>.amazonaws.com/enterprise-agent:v1
docker push <account>.dkr.ecr.<region>.amazonaws.com/enterprise-agent:v1

创建ECS Task Definition，配置环境变量（从Secrets Manager注入）

部署到ECS Service，配置Auto Scaling策略（CPU > 70% 扩容）

配置CloudWatch告警：响应时间 > 5s、错误率 > 1%

7.3 FastAPI服务化部署

核心优势：FastAPI提供高性能HTTP API接口，完美契合Agents-SDK的异步架构，支持自动文档生成、请求验证与生产级监控，是构建微服务化智能体系统的首选方案。

7.3.1 FastAPI应用架构

enterprise-agent-api/
├── app/
│   ├── __init__.py
│   ├── main.py                 # FastAPI应用入口
│   ├── api/
│   │   ├── __init__.py
│   │   ├── routes/
│   │   │   ├── __init__.py
│   │   │   └── agent.py        # 智能体API端点
│   │   └── dependencies.py     # 依赖注入
│   ├── core/
│   │   ├── __init__.py
│   │   └── config.py           # API配置
│   └── services/
│       ├── __init__.py
│       └── agent_service.py    # 智能体业务逻辑
├── Dockerfile
├── requirements.txt
└── .env

7.3.2 核心代码实现

# app/core/config.py
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    """应用配置"""
    OPENAI_API_KEY: str
    REDIS_URL: str = "redis://localhost:6379"
    API_VERSION: str = "v1"
    APP_NAME: str = "Enterprise Agent API"
    DEBUG: bool = False
    
    class Config:
        env_file = ".env"

settings = Settings()

# app/services/agent_service.py
import asyncio
from typing import List, Dict, Any
from agents import Agent, Runner
from app.core.config import settings
from utils.memory_store import RedisMemoryStore

class AgentService:
    """智能体服务层"""
    
    def __init__(self):
        self.memory_store = RedisMemoryStore(url=settings.REDIS_URL)
        
    async def process_message(
        self, 
        session_id: str, 
        message: str, 
        agent: Agent
    ) -> Dict[str, Any]:
        """处理单条消息"""
        # 加载历史记忆
        memory = await self.memory_store.load(session_id) or []
        
        # 执行智能体
        result = await Runner.run(agent, message, memory=memory)
        
        # 持久化记忆
        await self.memory_store.save(session_id, result.new_memory)
        
        return {
            "response": result.final_output,
            "session_id": session_id,
            "trace_id": result.trace.trace_id if result.trace else None
        }
    
    async def create_session(self) -> str:
        """创建新会话"""
        import uuid
        return str(uuid.uuid4())

# app/api/routes/agent.py
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel
from agents import Agent
from app.services.agent_service import AgentService
from agents.triage import create_triage_agent

router = APIRouter(prefix="/api/v1/agent", tags=["agent"])

class ChatRequest(BaseModel):
    """聊天请求模型"""
    message: str
    session_id: str | None = None

class ChatResponse(BaseModel):
    """聊天响应模型"""
    response: str
    session_id: str
    trace_id: str | None = None

# 依赖注入
def get_agent_service() -> AgentService:
    return AgentService()

def get_triage_agent() -> Agent:
    return create_triage_agent()

@router.post("/chat", response_model=ChatResponse)
async def chat(
    request: ChatRequest,
    agent_service: AgentService = Depends(get_agent_service),
    agent: Agent = Depends(get_triage_agent)
):
    """智能体对话接口"""
    try:
        # 创建新会话或复用现有会话
        session_id = request.session_id or await agent_service.create_session()
        
        # 处理消息
        result = await agent_service.process_message(
            session_id=session_id,
            message=request.message,
            agent=agent
        )
        
        return ChatResponse(**result)
        
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"智能体处理失败: {str(e)}")

@router.get("/health")
async def health_check():
    """健康检查端点"""
    return {"status": "healthy", "service": "agent-api"}

# app/main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.core.config import settings
from app.api.routes import agent

def create_application() -> FastAPI:
    """应用工厂函数"""
    app = FastAPI(
        title=settings.APP_NAME,
        version=settings.API_VERSION,
        debug=settings.DEBUG
    )
    
    # CORS配置
    app.add_middleware(
        CORSMiddleware,
        allow_origins=["*"],  # 生产环境应配置具体域名
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )
    
    # 注册路由
    app.include_router(agent.router)
    
    return app

app = create_application()

@app.get("/")
async def root():
    return {
        "message": "Enterprise Agent API",
        "docs": "/docs",
        "redoc": "/redoc"
    }

7.3.3 生产服务器配置

# requirements.txt 新增
fastapi>=0.115.0
uvicorn[standard]>=0.32.0
pydantic-settings>=2.0.0

# 启动命令（开发环境）
uvicorn app.main:app --reload --host 0.0.0.0 --port 8000

# 生产环境启动（Gunicorn + Uvicorn Workers）
gunicorn app.main:app -w 4 -k uvicorn.workers.UvicornWorker -b 0.0.0.0:8000

7.3.4 FastAPI专用Dockerfile

# Dockerfile.fastapi
FROM python:3.11-slim

WORKDIR /app

# 安装系统依赖
RUN apt-get update && apt-get install -y 
    gcc 
    && rm -rf /var/lib/apt/lists/*

# 复制依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 复制代码
COPY . .

# 非root用户
RUN useradd -m -u 1000 agentuser && chown -R agentuser:agentuser /app
USER agentuser

# 健康检查
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 
    CMD curl -f http://localhost:8000/api/v1/agent/health || exit 1

# 生产启动
CMD ["gunicorn", "app.main:app", "-w", "4", "-k", "uvicorn.workers.UvicornWorker", "-b", "0.0.0.0:8000"]

7.3.5 Kubernetes部署清单

# k8s-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: agent-api
spec:
  replicas: 3
  selector:
    matchLabels:
      app: agent-api
  template:
    metadata:
      labels:
        app: agent-api
    spec:
      containers:
      - name: agent-api
        image: <account>.dkr.ecr.<region>.amazonaws.com/enterprise-agent-api:v1
        ports:
        - containerPort: 8000
        env:
        - name: OPENAI_API_KEY
          valueFrom:
            secretKeyRef:
              name: agent-secrets
              key: openai-api-key
        - name: REDIS_URL
          value: "redis://redis-service:6379"
        resources:
          requests:
            memory: "512Mi"
            cpu: "250m"
          limits:
            memory: "1Gi"
            cpu: "500m"
        livenessProbe:
          httpGet:
            path: /api/v1/agent/health
            port: 8000
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /api/v1/agent/health
            port: 8000
          initialDelaySeconds: 5
          periodSeconds: 5

---
apiVersion: v1
kind: Service
metadata:
  name: agent-api-service
spec:
  type: LoadBalancer
  ports:
  - port: 80
    targetPort: 8000
  selector:
    app: agent-api

7.3.6 API调用示例

# 客户端调用示例
import requests

# 1. 首次调用（创建会话）
response = requests.post(
    "http://localhost:8000/api/v1/agent/chat",
    json={"message": "帮我查询订单123456789的物流"}
)
data = response.json()
print(f"响应: {data['response']}")
print(f"会话ID: {data['session_id']}")

# 2. 后续调用（携带会话ID）
session_id = data['session_id']
response = requests.post(
    "http://localhost:8000/api/v1/agent/chat",
    json={"message": "这个订单能退吗？", "session_id": session_id}
)
print(f"响应: {response.json()['response']}")

7.3.7 生产环境最佳实践

限流与防护

# 在app/main.py中添加
from fastapi_limiter import FastAPILimiter
import redis

@app.on_event("startup")
async def startup():
    redis_client = redis.from_url(settings.REDIS_URL, encoding="utf-8")
    await FastAPILimiter.init(redis_client)

监控指标

# 集成Prometheus
from prometheus_client import Counter, Histogram

request_latency = Histogram('agent_request_latency_seconds', '请求延迟')
request_count = Counter('agent_request_total', '总请求数', ['method', 'endpoint'])

@router.post("/chat")
async def chat(...):
    request_count.labels(method="POST", endpoint="/chat").inc()
    with request_latency.time():
        # ... 业务逻辑

链路追踪

# 集成OpenTelemetry
from opentelemetry import trace

tracer = trace.get_tracer(__name__)

@router.post("/chat")
async def chat(...):
    with tracer.start_as_current_span("agent_chat"):
        # ... 业务逻辑

---

八、核心最佳实践总结

8.1 开发规范

Instructions设计：采用编号列表，明确边界与流程，避免模糊指令

# ✅ 推荐
instructions="""1. 首先验证用户身份

然后查询订单信息

工具函数：

必须包含详细docstring使用Pydantic模型进行参数校验返回结构化字符串或JSON

记忆管理：始终使用 result.new_memory自动更新，禁止手动拼接消息列表

错误处理：

try:
    result = await Runner.run(agent, input, memory=memory)
except Exception as e:
    logger.error(f"智能体执行失败: {e}", exc_info=True)
    # 降级到备用智能体或返回友好提示

8.2 性能优化

模型选择：简单任务用 gpt-4o-mini，复杂推理用 gpt-4o，数学/代码用 o1Token控制：设置 max_tokens上限，监控token使用量异步并发：利用 asyncio.gather()并行执行独立任务缓存机制：对工具调用结果添加Redis缓存，设置TTL

8.3 安全与合规

密钥管理：绝不硬编码API Key，使用环境变量或云密钥服务权限隔离：每个智能体运行时传入 user_id，工具层验证权限数据脱敏：日志记录时自动过滤手机号、订单号等敏感信息审计追踪：启用OpenAI Traces，保存完整调用链路

8.4 学习资源

官方文档：platform.openai.com/docs/agents（持续更新）GitHub示例：github.com/openai/agents（官方示例库）社区论坛：OpenAI Developer Forum的Agents板块

---

版本声明：本笔记基于OpenAI Agents-SDK v0.5.0+ 编写，API可能随版本迭代调整，请以官方最新文档为准。