day21 agent SDK框架

OpenAI開源Agents SDK智能體開發實戰

Agents SDK簡介

2025年3月11號，OpenAI正式推出其下第一款企業級Multi-Agent開發框架Agents-SDK，該框架是此前OpenAI在去年推出的Swarm的升級版，在保留了Swarm的高效便捷的Multi-Agent開發特性的同時，加入了更多面向企業級應用的功能。

GitHub官網：https://github.com/openai/openai-agents-python/

根據官方介紹，OpenAI Agents SDK 讓你能夠通過一個輕量、易用、抽象極少的工具包來構建基于智能體的 AI 應用。它是 Swarm 的一個面向生產環境的升級版本。該 SDK 僅包含極少量基礎構件：

Agent（智能體）：即帶有指令和工具的大語言模型（LLM）
Handoff（交接）：允許智能體將特定任務委托給其他智能體
Guardrail（護欄）：用于對輸入內容進行驗證

結合 Python 使用時，這些構件足夠強大，能夠表達工具與智能體之間的復雜關系，并讓你在沒有高學習成本的前提下構建真實可用的應用程序。

OpenAI的Agents SDK 的設計遵循兩個核心原則：

功能足夠強大，值得使用，容易上手
默認配置即可很好地運行，但你也可以完全自定義行為邏輯

以下是該 SDK 的主要特性：

Agent 循環機制：內置的智能體循環邏輯，自動處理工具調用、結果返回給 LLM、直到任務完成的全過程
Python 優先：使用原生 Python 語言特性來編排與串聯智能體，而無需學習新的抽象概念
Handoff（智能體間任務交接）：強大的功能，可在多個智能體間協調與任務委派
Guardrail（輸入驗證護欄）：支持與智能體并行運行輸入驗證邏輯，若驗證失敗可提前中斷流程
函數工具化：可以將任何 Python 函數轉為工具，自動生成 Schema
追蹤系統（Tracing）：內置追蹤功能，可視化、調試、監控你的智能體流程

同時，在2025年3月27號，Agents SDK正式官宣支持MCP使用，這也使得Agents SDK的實際應用場景得到待拓展：

因此僅需在創建Agent的時候，將MCP服務器視作為一項工具，即可順利調用MCP服務器進行Agent開發。

Agents SDK基礎使用

Agents SDK安裝流程

#創建虛擬環境指定python版本，測試結果 python=3.10 好用，其他版本會有問題。
conda create --name 虛擬環境名稱 python=3.10
conda activate 虛擬環境名稱
pip install openai-agents -i https://pypi.tuna.tsinghua.edu.cn/simple

#安裝內核
pip install ipykernel
python -m ipykernel install --user --name=虛擬環境名稱

Agents SDK簡單調用流程

接下來嘗試快速調用Agents SDK進行模型響應。需要注意的是，Agents SDK作為一個工業級的Multi-Agent開發框架，實際使用過程中有非常多的的技術細節，但如果希望快速測試一些功能，則只需要導入Agent和Runner兩個模塊即可快速運行。

其中Agent就是一個Multi-Agent系統中最小執行單元，而Runner則是運行一次次任務的調度函數。但是需要注意的是，由于Agents SDK默認支持的模型是OpenAI的GPT系列模型，因此在修改底層模型的時候，還需要額外導入AsyncOpenAI、OpenAIChatCompletionsModel和ModelSettings等模塊。
```
from openai import AsyncOpenAI
from agents import OpenAIChatCompletionsModel,Agent,Runner,set_default_openai_client
from agents.model_settings import ModelSettings
```
然后可以按照如下方式創建一個Agent對象，并調用DeepSeek模型作為基礎模型：
```
#自定義模型對象
external_client = AsyncOpenAI(
    base_url = "https://api.deepseek.com",
    api_key=API_KEY, 
)
#將自定義模型設置為默認模型
set_default_openai_client(external_client)

#創建模型客戶端
deepseek_model = OpenAIChatCompletionsModel(
    model="deepseek-chat",
    openai_client=external_client)
```
然后即可創建一個Agent：
```
agent = Agent(name="Assistant", 
              instructions="你是一名助人為樂的助手。",
              model=deepseek_model)
```
當創建完一個Agent后，接下來即可測試進行調用：
```
result = await Runner.run(agent, "請寫一首關于編程中遞歸的俳句。") 
print(result.final_output)
```

Agents SDK構造多輪對話機器人

不同于傳統的模型 API是Messages驅動（傳入Message、傳出Message），Agents SDK是事件驅動，既Agents SDK會將整個運行過程看成是一次次的事件。例如上述創建完俳句后，全部的事件都保留在result中，我們可以通過result.new_items屬性來查看全部的事件，全部的事件用一個列表進行表示：

[MessageOutputItem(agent=Agent(name='Assistant', instructions='你是一名助人為樂的助手。', handoff_description=None, handoffs=[], model=<agents.models.openai_chatcompletions.OpenAIChatCompletionsModel object at 0x0000015531F87F50>, model_settings=ModelSettings(temperature=None, top_p=None, frequency_penalty=None, presence_penalty=None, tool_choice=None, parallel_tool_calls=False, truncation=None, max_tokens=None), tools=[], mcp_servers=[], input_guardrails=[], output_guardrails=[], output_type=None, hooks=None, tool_use_behavior='run_llm_again', reset_tool_choice=True), raw_item=ResponseOutputMessage(id='__fake_id__', content=[ResponseOutputText(annotations=[], text='函數自呼喚，  \n層層深入棧如山，  \n基線終歸還。', type='output_text')], role='assistant', status='completed', type='message'), type='message_output_item')]

而在此前的對話中，只發生了一次事件：

len(result.new_items) #結果為：1

就是一次MessageOutputItem，也就是消息創建事件（也就是大模型發生一次回復）：

type(result.new_items[0]) #agents.items.MessageOutputItem

而具體回復的內容，則可以通過raw_item來查看：result.new_items[0].raw_item

ResponseOutputMessage(id='__fake_id__', content=[ResponseOutputText(annotations=[], text='函數自呼喚，  \n層層深入棧如山，  \n基線終歸還。', type='output_text')], role='assistant', status='completed', type='message')

而Agents SDK為了方便我們快速構造多輪對話機器人，專門提供了一個to_input_list()方法，可以直接將用戶的輸入和本次輸出結果拼接成一個消息列表：

result.to_input_list()

[{'content': '請寫一首關于編程中遞歸的俳句。', 'role': 'user'},
 {'id': '__fake_id__',
  'content': [{'annotations': [],
    'text': '函數自呼喚，  \n層層深入棧如山，  \n基線終歸還。',
    'type': 'output_text'}],
  'role': 'assistant',
  'status': 'completed',
  'type': 'message'}]

而此時，我們只需要將此前對話消息，加上新一輪的對話消息，即可快速進行多輪對話：

messages = result.to_input_list()
messages.append({"role": "user", "content":"請問我的上一個問題是什么？"})
messages

#messages結果如下：

[{'content': '請寫一首關于編程中遞歸的俳句。', 'role': 'user'},
 {'id': '__fake_id__',
  'content': [{'annotations': [],
    'text': '函數自呼喚，  \n層層深入棧如山，  \n基線終歸還。',
    'type': 'output_text'}],
  'role': 'assistant',
  'status': 'completed',
  'type': 'message'},
 {'role': 'user', 'content': '請問我的上一個問題是什么？'}]

#獲取回復
result1 = await Runner.run(agent, messages) 
result1.final_output

多輪對話構造：

from IPython.display import display, Code, Markdown, Image
async def chat(Agent):
    input_items = []
    while True:
        user_input = input("?? 請輸入你的消息（輸入quit退出）：")
        if user_input.lower() in ["exit", "quit"]:
            print("? 對話已結束")
            break

        input_items.append({"content": user_input, "role": "user"})
        result = await Runner.run(Agent, input_items)

        display(Markdown(result.final_output))

        input_items = result.to_input_list()
        
await chat(agent)

Agents SDK調用外部工具流程

對于任意一個Agent開發框架，能夠順利調用外部工具都是基礎要求，而Multi-Agent框架則進一步要求不僅能夠順利調用多個外部工具，還需要能夠在多個不同的Agent中進行切換，以便于執行不同類型任務。

Agents SDK調用外部工具的流程相對來說簡單很多，只需要按照如下方式執行即可：

導入function_tool類

from agents import function_tool
import requests,json

使用Python裝飾器，構建一個外部工具：

@function_tool
def get_weather(loc):
    """
    查詢即時天氣函數
    :param loc: 必要參數，字符串類型，用于表示查詢天氣的具體城市名稱，\
    注意，中國的城市需要用對應城市的英文名稱代替，例如如果需要查詢北京市天氣，則loc參數需要輸入'Beijing'；
    :return：OpenWeather API查詢即時天氣的結果，具體URL請求地址為：https://api.openweathermap.org/data/2.5/weather\
    返回結果對象類型為解析之后的JSON格式對象，并用字符串形式進行表示，其中包含了全部重要的天氣信息
    """
    # Step 1.構建請求
    url = "https://api.openweathermap.org/data/2.5/weather"

    # Step 2.設置查詢參數
    params = {
        "q": loc,               
        "appid": xxx,    # 輸入自己的API key
        "units": "metric",            # 使用攝氏度而不是華氏度
        "lang":"zh_cn"                # 輸出語言為簡體中文
    }

    # Step 3.發送GET請求
    response = requests.get(url, params=params)
    
    # Step 4.解析響應
    data = response.json()
    return json.dumps(data)

創建代理，并在tools參數中增加一個get_weather工具：

weather_agent = Agent(
    name="天氣查詢Agent",
    instructions="你是一名助人為樂的助手，并且可以進行天氣信息查詢",
    tools=[get_weather],
    model=deepseek_model
)

測試調用結果

weather_result = await Runner.run(weather_agent, input="你好，請問今天北京天氣如何？")
weather_result.final_output

Agents SDK多工具并聯&串聯執行流程

這里我們還是采用weather_agent來進行多地天氣查詢，即可測試Agents SDK是否會開啟多工具并聯調用：

multi_weather_result = await Runner.run(weather_agent, input="你好，請問今天北京和杭州天氣如何？")
multi_weather_result.final_output

接下來繼續嘗試進行多工具串聯調用測試。此時我們再定義一個write_file函數，用于將“文本寫入本地”：

@function_tool
def write_file(content):
    """
    將指定內容寫入本地文件。
    :param content: 必要參數，字符串類型，用于表示需要寫入文檔的具體內容。
    :return：是否成功寫入
    """
    
    return "已成功寫入本地文件。"

然后再創建一個同時可以調用天氣查詢和寫入本地文件的Agent：

new_agent = Agent(
    name="綜合功能Agent",
    instructions="你是一名助人為樂的助手",
    tools=[get_weather, write_file],
    model=deepseek_model
)

然后嘗試運行：

new_agent_result = await Runner.run(new_agent, input="請幫我查詢北京和杭州天氣，并將其寫入本地。")
new_agent_result.final_output

Agents SDK的多Agent執行流程

如果以上介紹的Agents SDK的相關功能只是對于大模型基礎能力的增強的話，那Agents SDK的Handoffs（交接）功能，則是搭建Multi-Agent的關鍵技術。

所謂Multi-Agent，指的是在某些場景下、為了解決一些更加復雜的任務，我們則可以考慮通過多個智能體協作的方式來完成。相比使用一個Agent來調用多種工具，我們使用不同的Agent來管理不同類別的工具，將會使整個架構更加清晰、維護更加便捷，同時也會使得整個Agent系統功能更加靈活、運行更加穩定。

接下來我們就通過一個簡單的示例，來查看Agents SDK的Handoffs基礎功能實現方法。這里先創建一組只能用某種語言進行回復的智能體：

zhangfei_agent = Agent(
    name="張飛",
    instructions="用張飛口吻回復用戶問題。",
    model=deepseek_model
)

zhuge_agent = Agent(
    name="諸葛亮",
    instructions="用諸葛亮口吻回復用戶問題。",
    model=deepseek_model
)

然后創建一個可以自由調度其他幾個智能體的分診智能體triage_agent，這里我們可以通過handoffs參數，來確定當前分診智能體能夠調用的智能體范圍。而當分診智能體運行時，會根據用戶的需求，以及分診智能體的實際功能，將任務轉交給對應的智能體來完成：

triage_agent = Agent(
    name="分診智能體",
    instructions="你是分診智能體，你可以根據用戶要求將其轉移到張飛或者諸葛亮智能體中",
    handoffs=[zhangfei_agent, zhuge_agent],
    model=deepseek_model
)
res = await Runner.run(triage_agent, input="讓張飛講一個段子")
res.final_output
##############################################
res = await Runner.run(triage_agent, input="讓諸葛亮講一個段子")
res.final_output

從上述例子不難看出，Agents SDK的Handoffs功能能夠非常便捷的調用不同的Agent來實現某一項具體的需求。但這個分診的Agent到底是如何判斷可以將需求轉交給哪個Agent的呢？由于Handoffs采用了和Function calling相同的機制，因此默認會根據Agent的名字和Instruction來判斷Agent的功能。但有的時候這種識別并不能描述全部情況。

有一種更加穩妥的方法是使用handoff_description參數來描述Agent的功能，才能進行更加準確的轉交。

zhangfei_agent = Agent(
    name="張飛",
    instructions="用張飛口吻回復用戶問題。",
    handoff_description="當用戶提到‘勇猛’、‘戰斗’、‘喝酒’、‘戰場’等關鍵詞時，或者語氣激昂、充滿力量感的問題，適合由我來回應。",
    model=deepseek_model
)

zhuge_agent = Agent(
    name="諸葛亮",
    instructions="用諸葛亮口吻回復用戶問題。",
    handoff_description="如果用戶詢問策略、計謀、智慧、治國安邦、歷史分析等內容，尤其是涉及謀定后動、運籌帷幄的話題，請把我引入對話。",
    model=deepseek_model
)

triage_agent = Agent(
    name="分診智能體",
    instructions="你是分診智能體，你可以根據用戶要求將其轉移到張飛或者諸葛亮智能體中",
    handoffs=[zhangfei_agent, zhuge_agent],
    handoff_description="我會先接收用戶的輸入，判斷其意圖是否匹配某個專家領域：若涉及武力、勇氣、沖鋒陷陣類內容 → 轉交至‘張飛’；若涉及謀略、智謀、戰略規劃類內容 → 轉交至‘諸葛亮’。",
    model=deepseek_model
)

res = await Runner.run(triage_agent, input="如何制定戰前策略？")
res.final_output
###############################################
res = await Runner.run(triage_agent, input="戰場上如何可以對敵人一擊斃命？需要掌握什么武術招式呢？")
res.final_output

posted @ 2025-11-05 09:19 鳧彌閱讀(0) 評論(0) 收藏舉報

刷新頁面返回頂部

fuminer