執行代理(agents)
你可以透過 [Runner
][agents.run.Runner] 類別來執行代理。你有三種選擇:
- [
Runner.run()
][agents.run.Runner.run],這是一個非同步(async)方法,會回傳一個 [RunResult
][agents.result.RunResult]。 - [
Runner.run_sync()
][agents.run.Runner.run_sync],這是一個同步(sync)方法,實際上是在底層執行.run()
。 - [
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 會執行以下循環流程:
- 我們會針對當前代理,使用目前的輸入呼叫大型語言模型 (LLM)。
- 大型語言模型 (LLM) 產生其輸出。
- 如果 LLM 回傳
final_output
,則循環結束並回傳結果。 - 如果 LLM 進行 handoff,我們會更新當前代理與輸入,然後重新執行循環。
- 如果 LLM 產生工具呼叫(tool calls),我們會執行這些工具呼叫,將結果附加後,重新執行循環。
- 如果 LLM 回傳
- 如果超過所傳入的
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) 專屬設定。例如,你可以設定全域temperature
或top_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 呼叫),但這代表聊天對話中的單一邏輯回合。例如:
- 使用者回合:使用者輸入文字
- 執行器 (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.run
、Runner.run_sync
或Runner.run_streamed
方法)時,會拋出此例外。這表示代理無法在指定的互動輪數內完成任務。 - [
ModelBehaviorError
][agents.exceptions.ModelBehaviorError]:當底層模型(大型語言模型,LLM)產生非預期或無效輸出時,會發生此例外。常見情境包括:- 格式錯誤的 JSON:當模型針對工具呼叫或直接輸出時,提供了格式錯誤的 JSON 結構,特別是在有明確定義
output_type
的情況下。 - 非預期的工具相關失敗:模型未以預期方式使用工具時
- 格式錯誤的 JSON:當模型針對工具呼叫或直接輸出時,提供了格式錯誤的 JSON 結構,特別是在有明確定義
- [
UserError
][agents.exceptions.UserError]:當你(使用 SDK 撰寫程式的人)在使用 SDK 時發生錯誤時,會拋出此例外。通常是因為程式實作錯誤、設定無效或 API 誤用所導致。 - [
InputGuardrailTripwireTriggered
][agents.exceptions.InputGuardrailTripwireTriggered]、[OutputGuardrailTripwireTriggered
][agents.exceptions.OutputGuardrailTripwireTriggered]:當輸入或輸出 guardrail 條件被觸發時,分別會拋出這些例外。輸入 guardrail 會在處理前檢查進來的訊息,輸出 guardrail 則會在回應送出前檢查代理的最終回應。