Skip to content

執行代理(agents)

你可以透過 [Runner][agents.run.Runner] 類別來執行代理。你有三種選擇:

  1. [Runner.run()][agents.run.Runner.run],這是一個非同步(async)方法,會回傳一個 [RunResult][agents.result.RunResult]。
  2. [Runner.run_sync()][agents.run.Runner.run_sync],這是一個同步(sync)方法,實際上是在底層執行 .run()
  3. [Runner.run_streamed()][agents.run.Runner.run_streamed],這是一個非同步(async)方法,會回傳一個 [RunResultStreaming][agents.result.RunResultStreaming]。它會以串流模式呼叫大型語言模型(LLM),並在接收到事件時即時將這些事件串流給你。
from agents import Agent, Runner

async def main():
    agent = Agent(name="Assistant", instructions="You are a helpful assistant")

    result = await Runner.run(agent, "Write a haiku about recursion in programming.")
    print(result.final_output)
    # Code within the code,
    # Functions calling themselves,
    # Infinite loop's dance

閱讀更多內容,請參見 results guide

代理循環(agent loop)

當你在 Runner 中使用 run 方法時,你需要傳入一個起始代理(starting agent)以及輸入。輸入可以是一個字串(視為使用者訊息),也可以是一個輸入項目列表,這些項目即為 OpenAI Responses API 中的項目。

runner 會執行以下循環流程:

  1. 我們會針對當前代理,使用目前的輸入呼叫大型語言模型 (LLM)。
  2. 大型語言模型 (LLM) 產生其輸出。
    1. 如果 LLM 回傳 final_output,則循環結束並回傳結果。
    2. 如果 LLM 進行 handoff,我們會更新當前代理與輸入,然後重新執行循環。
    3. 如果 LLM 產生工具呼叫(tool calls),我們會執行這些工具呼叫,將結果附加後,重新執行循環。
  3. 如果超過所傳入的 max_turns,我們會拋出 [MaxTurnsExceeded][agents.exceptions.MaxTurnsExceeded] 例外。

Note

判斷大型語言模型 (LLM) 輸出是否為「最終輸出」的規則是:當它產生所需型別的文字輸出,且沒有工具呼叫時,即視為最終輸出。

串流 (Streaming)

串流 (Streaming) 允許你在 LLM 執行時額外接收串流事件。當串流結束後,[RunResultStreaming][agents.result.RunResultStreaming] 會包含關於這次執行的完整資訊,包括所有新產生的輸出。你可以呼叫 .stream_events() 來取得串流事件。詳細內容請參閱 streaming guide

執行設定 (Run config)

run_config 參數可讓你設定代理 (agent) 執行的一些全域設定:

  • [model][agents.run.RunConfig.model]:允許設定全域 LLM 模型,不論每個代理 (Agent) 的 model 為何。
  • [model_provider][agents.run.RunConfig.model_provider]:模型提供者,用於查找模型名稱,預設為 OpenAI。
  • [model_settings][agents.run.RunConfig.model_settings]:覆蓋代理 (agent) 專屬設定。例如,你可以設定全域 temperaturetop_p
  • [input_guardrails][agents.run.RunConfig.input_guardrails]、[output_guardrails][agents.run.RunConfig.output_guardrails]:在所有執行中加入的輸入或輸出 guardrails 清單。
  • [handoff_input_filter][agents.run.RunConfig.handoff_input_filter]:全域輸入過濾器,會套用到所有 handoff,如果 handoff 尚未有設定的話。輸入過濾器允許你編輯傳送給新代理 (agent) 的輸入。更多細節請參閱 [Handoff.input_filter][agents.handoffs.Handoff.input_filter] 的文件說明。
  • [tracing_disabled][agents.run.RunConfig.tracing_disabled]:允許你停用整個執行過程的 tracing
  • [trace_include_sensitive_data][agents.run.RunConfig.trace_include_sensitive_data]:設定追蹤 (trace) 是否包含可能的敏感資料,例如 LLM 及工具呼叫的輸入/輸出。
  • [workflow_name][agents.run.RunConfig.workflow_name]、[trace_id][agents.run.RunConfig.trace_id]、[group_id][agents.run.RunConfig.group_id]:設定此次執行的追蹤工作流程名稱、追蹤 ID 與追蹤群組 ID。我們建議至少設定 workflow_name。群組 ID 為選填欄位,可讓你跨多次執行關聯追蹤。
  • [trace_metadata][agents.run.RunConfig.trace_metadata]:要在所有追蹤中包含的中繼資料。

對話/聊天執行緒 (Conversations/chat threads)

呼叫任一執行方法 (run methods) 可能會導致一個或多個代理 (agent) 執行(因此也可能有一個或多個 LLM 呼叫),但這代表聊天對話中的單一邏輯回合。例如:

  1. 使用者回合:使用者輸入文字
  2. 執行器 (Runner) 執行:第一個代理 (agent) 呼叫 LLM、執行工具、handoff 給第二個代理 (agent),第二個代理再執行更多工具,最後產生輸出。

在代理 (agent) 執行結束時,你可以選擇要顯示哪些內容給使用者。例如,你可以顯示所有代理 (agent) 新產生的項目,或只顯示最終輸出。無論哪種方式,使用者都可能接著提出後續問題,此時你可以再次呼叫執行方法。

手動對話管理

你可以使用 [RunResultBase.to_input_list()][agents.result.RunResultBase.to_input_list] 方法手動管理對話歷史,取得下一回合的輸入:

async def main():
    agent = Agent(name="Assistant", instructions="Reply very concisely.")

    thread_id = "thread_123"  # Example thread ID
    with trace(workflow_name="Conversation", group_id=thread_id):
        # First turn
        result = await Runner.run(agent, "What city is the Golden Gate Bridge in?")
        print(result.final_output)
        # San Francisco

        # Second turn
        new_input = result.to_input_list() + [{"role": "user", "content": "What state is it in?"}]
        result = await Runner.run(agent, new_input)
        print(result.final_output)
        # California

使用 Sessions 進行自動化對話管理

如果你想採用更簡單的方法,可以使用 Sessions,自動處理對話歷史,無需手動呼叫 .to_input_list()

from agents import Agent, Runner, SQLiteSession

async def main():
    agent = Agent(name="Assistant", instructions="Reply very concisely.")

    # Create session instance
    session = SQLiteSession("conversation_123")

    thread_id = "thread_123"  # Example thread ID
    with trace(workflow_name="Conversation", group_id=thread_id):
        # First turn
        result = await Runner.run(agent, "What city is the Golden Gate Bridge in?", session=session)
        print(result.final_output)
        # San Francisco

        # Second turn - agent automatically remembers previous context
        result = await Runner.run(agent, "What state is it in?", session=session)
        print(result.final_output)
        # California

Sessions 會自動:

  • 在每次執行前擷取對話歷史紀錄
  • 在每次執行後儲存新訊息
  • 針對不同的 session ID 維護獨立的對話

詳細資訊請參閱 Sessions documentation

長時間執行代理 (Long running agents) 與人類參與 (human-in-the-loop)

你可以使用 Agents SDK 的 Temporal 整合,來執行具備持久性、可長時間運作的工作流程,包括需要人類參與的任務。你可以在 這段影片 中觀看 Temporal 與 Agents SDK 協同完成長時間任務的實際示範,並可在此查看相關文件

例外狀況 (Exceptions)

SDK 在特定情況下會拋出例外。完整清單請參見 [agents.exceptions][]。以下為概述:

  • [AgentsException][agents.exceptions.AgentsException]:這是 SDK 內所有例外的基礎類別。它作為通用型別,所有其他特定例外皆從此類別衍生。
  • [MaxTurnsExceeded][agents.exceptions.MaxTurnsExceeded]:當代理的執行超過傳入 max_turns 限制(用於 Runner.runRunner.run_syncRunner.run_streamed 方法)時,會拋出此例外。這表示代理無法在指定的互動輪數內完成任務。
  • [ModelBehaviorError][agents.exceptions.ModelBehaviorError]:當底層模型(大型語言模型,LLM)產生非預期或無效輸出時,會發生此例外。常見情境包括:
    • 格式錯誤的 JSON:當模型針對工具呼叫或直接輸出時,提供了格式錯誤的 JSON 結構,特別是在有明確定義 output_type 的情況下。
    • 非預期的工具相關失敗:模型未以預期方式使用工具時
  • [UserError][agents.exceptions.UserError]:當你(使用 SDK 撰寫程式的人)在使用 SDK 時發生錯誤時,會拋出此例外。通常是因為程式實作錯誤、設定無效或 API 誤用所導致。
  • [InputGuardrailTripwireTriggered][agents.exceptions.InputGuardrailTripwireTriggered]、[OutputGuardrailTripwireTriggered][agents.exceptions.OutputGuardrailTripwireTriggered]:當輸入或輸出 guardrail 條件被觸發時,分別會拋出這些例外。輸入 guardrail 會在處理前檢查進來的訊息,輸出 guardrail 則會在回應送出前檢查代理的最終回應。