分类： Agent

核心概念

1. 智能体（Agents）：配置了指令、工具、防护栏和移交功能的LLM
2. 移交（Handoffs）：允许智能体将控制权转移给其他智能体来处理特定任务
3. 防护栏（Guardrails）：可配置的输入/输出验证安全检查
4. 追踪（Tracing）：内置的智能体运行跟踪，用于调试和优化

安装

git clone https://github.com/openai/openai-agents-python.git
cd openai-agents-python

设置Python环境：

python -m venv env
source env/bin/activate  # Linux/Mac
# 或者在Windows上：
# env\Scripts\activate

安装SDK：

pip install openai-agents

设置API密钥：

# Linux/Mac
export OPENAI_API_KEY=你的密钥

# Windows
# set OPENAI_API_KEY=你的密钥

运行示例：

# 例如，运行hello world示例
cd examples/basic
python hello_world.py

实例分析：

• agent_patterns文件夹

agents_as_tools.py：

• 展示了”代理即工具”模式
• 前线代理（orchestrator_agent）接收用户消息，然后从一组翻译代理中选择合适的工具
• 包含西班牙语、法语和意大利语翻译代理
• 最后使用合成代理检查翻译并产生最终响应

deterministic.py：

• 展示了确定性流程，每个步骤由特定代理执行
• 流程：生成故事大纲 → 检查大纲质量和类型 → 如果符合条件则创作故事
• 包含质量门控检查，如果大纲不是高质量或不是科幻故事则终止流程

input_guardrails.py：

• 展示了如何使用输入防护机制
• 防护机制在代理执行的同时并行运行
• 实现了一个检测用户是否请求解决数学作业的防护措施
• 如果防护触发，会用拒绝消息响应而不是正常处理请求

llm_as_a_judge.py：

• 展示了”LLM作为评判者”模式
• 第一个代理生成故事大纲，第二个代理评判并提供反馈
• 循环迭代直到评判代理对大纲满意为止
• 强制要求至少一次修改（”第一次尝试永远不给通过”）

output_guardrails.py：

• 展示了如何使用输出防护机制
• 检查代理输出是否包含敏感数据（这里是电话号码）
• 如果检测到敏感数据，防护措施会触发

parallelization.py：

• 展示了并行化模式
• 并行运行同一代理三次，产生三个不同的翻译结果
• 然后使用选择代理挑选最佳翻译

routing.py：

• 展示了交接/路由模式
• 分流代理接收第一条消息，然后根据请求的语言交给合适的代理
• 支持法语、西班牙语和英语代理
• 响应实时流式传输给用户

hello_world.py

这是最基本的例子，展示了如何创建一个简单的代理并运行它：

• 创建一个只用海口(haiku)回答的代理
• 使用 Runner.run() 执行代理
• 获取最终输出结果

dynamic_system_prompt.py

展示了如何基于上下文动态设置代理的指令：

• 创建自定义上下文类(CustomContext)，包含代理的风格
• 定义动态指令函数，根据风格返回不同的指令
• 随机选择风格(海口、海盗或机器人)并运行代理

agent_lifecycle_example.py 和 lifecycle_example.py

这两个文件展示了代理生命周期的监控：

• 定义自定义钩子类来监控代理的各种事件
• 使用工具(如 random_number 和 multiply_by_two)
• 设置代理之间的交接逻辑(如果数字为奇数则交给乘法代理)
• 监控和输出代理执行的完整生命周期

区别在于 agent_lifecycle_example.py 使用 AgentHooks 绑定到特定代理，而 lifecycle_example.py 使用 RunHooks 绑定到整个运行过程。

stream_items.py 和 stream_text.py

展示了框架的流式输出功能：

• stream_items.py 展示如何处理各种流式事件(工具调用、工具输出、消息输出等)
• stream_text.py 展示如何直接流式输出文本内容

计算机控制：通过 ComputerTool 和 AsyncComputer 接口，代理可以直接控制浏览器执行复杂的网页操作任务，这对于自动化网络交互、数据收集和UI测试非常有用。

向量搜索：通过 FileSearchTool，代理可以搜索预先准备的向量数据库，适用于文档检索、知识库问答等场景。

网络搜索：通过 WebSearchTool，代理可以直接搜索互联网获取实时信息，适合需要最新数据的场景。

演示

research_bot

web_search.py

微信：adoresever

gemini进行text2SQL的查询

from smolagents import CodeAgent
from smolagents import tool, LiteLLMModel
from sqlalchemy import create_engine, MetaData, Table, Column, String, Integer, Float, insert, text

engine = create_engine("sqlite:///:memory:")
metadata = MetaData()

products = Table(
    "products",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("name", String(50)),
    Column("category", String(20)),
    Column("price", Float),
    Column("stock", Integer)
)

sales = Table(
    "sales",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("product_id", Integer),
    Column("quantity", Integer),
    Column("sale_date", String(10))
)

metadata.create_all(engine)

# 示例数据
product_data = [
    {"id": 1, "name": "游戏本", "category": "电脑", "price": 6999.0, "stock": 100},
    {"id": 2, "name": "机械键盘", "category": "配件", "price": 299.0, "stock": 50},
    {"id": 3, "name": "游戏手柄", "category": "配件", "price": 199.0, "stock": 30},
    {"id": 4, "name": "办公本", "category": "电脑", "price": 4999.0, "stock": 80}
]

sales_data = [
    {"id": 1, "product_id": 1, "quantity": 2, "sale_date": "2024-01-01"},
    {"id": 2, "product_id": 2, "quantity": 5, "sale_date": "2024-01-02"},
    {"id": 3, "product_id": 1, "quantity": 1, "sale_date": "2024-01-03"},
    {"id": 4, "product_id": 4, "quantity": 3, "sale_date": "2024-01-03"}
]

with engine.begin() as conn:
    for item in product_data:
        conn.execute(insert(products).values(item))
    for item in sales_data:
        conn.execute(insert(sales).values(item))

@tool
def sql_engine(query: str) -> str:
    """执行SQL查询。

    Args:
        query: SQL查询语句

    Returns:
        str: 查询结果
    """
    try:
        with engine.connect() as conn:
            result = conn.execute(text(query))
            columns = result.keys()
            rows = result.fetchall()
            
            if not rows:
                return "查询没有返回任何结果"

            output = []
            output.append(" | ".join(str(col) for col in columns))
            output.append("-" * (sum(len(str(col)) for col in columns) + 3 * (len(columns) - 1)))
            
            for row in rows:
                output.append(" | ".join(str(val) for val in row))
                
            return "\n".join(output)
    except Exception as e:
        return f"SQL执行错误: {str(e)}"

model = LiteLLMModel(model_id="gemini/gemini-2.0-flash-exp")
#ollama/qwen2.5:14b
agent = CodeAgent(
    tools=[sql_engine],
    model=model,
    verbose=True
)

test_query = "请查找库存量最多的三种商品"
print("执行查询:", test_query)
result = agent.run(test_query)
print("查询结果:")
print(result)

ollama模型调用duckduckgo进行网络查询

from smolagents import CodeAgent, DuckDuckGoSearchTool  
from smolagents import tool, TransformersModel, LiteLLMModel
from typing import Optional

model = LiteLLMModel(
    model_id="ollama/qwen2.5:14b",  # 使用 Ollama 格式的模型 ID
    api_base="http://localhost:11434"  # Ollama 的本地地址
)

# 创建 Agent 实例
agent = CodeAgent(
    tools=[DuckDuckGoSearchTool()],
    model=model,
    verbose=True  
)

# 运行查询
print(agent.run("中国第六代战机"))