OpenAI Agents SDK 与 ModelRiver 并驱

让由 OpenAI 官方御马监监造出产的正牌代理框架交由 ModelRiver 之手去做大调派。让每一步代理动作都受泽深厚从而享有无拘束跨厂牌代执行作业、遇死机自动切档重开还有事无巨细皆能可供盘问观测探勘的安全结界。

概况一览 (Overview)

这卷出自 OpenAI Agents SDK 大册的法门乃是一通身无赘肉极为干练、可被直接端上来推向前线开火压战阵用以打造带有抄拿趁手工具、会跟兄弟单位递折子走马换将(handoffs)并有输入/输出边界安全防脱轨框(guardrails)的独立行走自主机器干探大架构。若将其拨转枪头向着 ModelRiver 输诚交令,那么手底下的干员探头们便再也不会被栓生死只认准门外 OpenAI 这一家厂牌门庭了:不仅什么牌子货都能使唤甚至还能自带后援梯队出场时自行无痛大挪移接防(failover)。

把这套玩明白后您赢取的确确切切筹码:

  • 驾着属于原本 Agents SDK 的这挂牛车照旧可以向谁家(OpenAI, Anthropic, Mistral 也好全不拘着)都能收单开拔叫战
  • 行进走线于曲折多步环圈回套代理拉锯里一旦谁趴窝死机自有救生大替补不惊扰任何人立刻暗箱接单续命
  • 探子走错每一拍每一阶消耗之花项均立等可见于账目榜 盘查记事册(Request Logs) 分文不少
  • 当需换手换边更调换场家出勤时动动这端面上之按钮盘便定死敲死了压根免除了要去源码堆翻找去再动土之烦碍

快点上手 (Quick start)

把大包带料拉齐活

Bash
pip install openai-agents

给它接死在 ModelRiver 大闸上口

PYTHON
1from agents import Agent, Runner
2from openai import AsyncOpenAI
3 
4client = AsyncOpenAI(
5 base_url="https://api.modelriver.com/v1",
6 api_key="mr_live_YOUR_API_KEY",
7)
8 
9agent = Agent(
10 name="Assistant",
11 instructions="从今天起你就是位时刻乐意排忧解难替人作答清楚明了的随侍跟班了。",
12 model="my-chat-workflow",
13)
14 
15result = Runner.run_sync(agent, "敢问这头顶的 ModelRiver 是尊作何用处的金佛?", client=client)
16print(result.final_output)

腰间别好随取随用大兵刃杂项五金工具的出勤密探 (Agent with tools)

PYTHON
1from agents import Agent, Runner, function_tool
2from openai import AsyncOpenAI
3 
4client = AsyncOpenAI(
5 base_url="https://api.modelriver.com/v1",
6 api_key="mr_live_YOUR_API_KEY",
7)
8 
9@function_tool
10def get_weather(location: str) -> str:
11 """替人四处打探当下的云天日头作何动向气候。"""
12 return f"据报眼下身在 {location} 此地室外为 22°C 还挂着老大的太阳"
13 
14@function_tool
15def search_knowledge_base(query: str) -> str:
16 """去探囊取物搜罗藏于自家后院库房门内的档书残卷。"""
17 return f"翻得一页:所谓这 ModelRiver 者实乃一专职充任路由中正中枢专门给跨厂大 AI 算力批条派单的门户之地。"
18 
19agent = Agent(
20 name="Research Assistant",
21 instructions="去给人解惑找找经卷资料去。碰到用得上的趁手家什你别客气尽管掏。",
22 model="my-chat-workflow",
23 tools=[get_weather, search_knowledge_base],
24)
25 
26result = Runner.run_sync(agent, "有人晓得东京上头那片天是个甚么动静没?", client=client)
27print(result.final_output)

各司其职分掌大印干探转档交班大法 (Agent handoffs)

这就去拼出一席大盘能任由里边那干人将个大块麻烦差事化整为零丢包指手给分得更细门派掌门去的联通法阵:

PYTHON
1from agents import Agent, Runner
2from openai import AsyncOpenAI
3 
4client = AsyncOpenAI(
5 base_url="https://api.modelriver.com/v1",
6 api_key="mr_live_YOUR_API_KEY",
7)
8 
9# 各门专门坐镇各方山头只挑刺各自能耐的术业有专攻高阶探将
10billing_agent = Agent(
11 name="Billing Specialist",
12 instructions="归你管拿捏这些讨钱付现要结单的烦心开支账事项。务须要精准到角分计价没商量。",
13 model="my-chat-workflow",
14)
15 
16tech_agent = Agent(
17 name="Tech Support",
18 instructions="所有那沾着生僻大硬头机器技术原理难题的锅你背。务须要有板有眼交出带码盘演示之样例真章。",
19 model="my-chat-workflow",
20)
21 
22# 前厅这方专门负责望风发包挑担派活的前台打杂把门小官
23triage_agent = Agent(
24 name="Triage",
25 instructions="把那瞎嚷嚷的人客诉求按方抓药归拢理出门道向后给递条子让对门能应这事的人出来平事。",
26 model="my-chat-workflow",
27 handoffs=[billing_agent, tech_agent],
28)
29 
30result = Runner.run_sync(triage_agent, "请问若是升了那叫 Pro 的档位计费得砸进去多少门票钱?", client=client)
31print(result.final_output)

带锁链防发疯跳栏杆拉底线定身术符咒防暴制动环 (Guardrails)

额外给注入加上关于输入输出的严查审过大阀:

PYTHON
1from agents import Agent, Runner, InputGuardrail, OutputGuardrail, GuardrailFunctionOutput
2from pydantic import BaseModel
3 
4class SafetyCheck(BaseModel):
5 is_safe: bool
6 reason: str
7 
8async def check_input(ctx, agent, input_text):
9 # 用个疾走如风耗资极廉的车辙道去搞定这个单纯打大检查哨安全甄别的勾当
10 result = await Runner.run(
11 Agent(
12 name="Safety Checker",
13 instructions="死死盯牢并排定查验交来的这票文墨字货正经端庄否。若眼见其带邪毒涉暗箭只管将假 (is_safe=false) 印鉴戳它头上。",
14 model="fast-safety-check", # 这门用个路费起步极廉价又跑的飞起的便宜小槽通即可
15 output_type=SafetyCheck,
16 ),
17 input_text,
18 context=ctx,
19 )
20 return GuardrailFunctionOutput(
21 output_info=result.final_output,
22 tripwire_triggered=not result.final_output.is_safe,
23 )
24 
25agent = Agent(
26 name="Guarded Assistant",
27 instructions="从生在此刻起,你就得做个常相伴的好心助阵随员了。",
28 model="my-chat-workflow",
29 input_guardrails=[InputGuardrail(guardrail_function=check_input)],
30)

潺流泻地成行挂瀑下泄之连击走流 (Streaming)

PYTHON
1import asyncio
2from agents import Agent, Runner
3from openai import AsyncOpenAI
4 
5client = AsyncOpenAI(
6 base_url="https://api.modelriver.com/v1",
7 api_key="mr_live_YOUR_API_KEY",
8)
9 
10agent = Agent(
11 name="Storyteller",
12 instructions="只管天马行空给人攒那种扯得让人拍大腿叫绝带沉浸式代入的大好故事篇。",
13 model="my-chat-workflow",
14)
15 
16async def main():
17 result = Runner.run_streamed(agent, "跟俺胡扯一遭段子里头有个叫人工智能物主干出啥好事的大本子事罢", client=client)
18 async for event in result.stream_events():
19 if event.type == "raw_response_event" and hasattr(event.data, "delta"):
20 print(event.data.delta, end="", flush=True)
21 
22asyncio.run(main())

依着探员自身面目特性因地适宜换型号分包打挂工作流切道分行

分封指定差使时就让各有绝活的探子进专门对口能搭能造的 ModelRiver 工作大特列管:

PYTHON
1# 给跑街站前台分包的干员配那最不讲理生猛要快字排头的小摩托 (速度才是真章)
2triage_agent = Agent(name="Triage", model="fast-triage", ...)
3 
4# 请那种胸中大蕴学海深测不见底的主咖接应并供足重火器解决奇难杂症
5expert_agent = Agent(name="Expert", model="deep-expert", ...)
6 
7# 让做边卫守规线那查抄关哨只图价廉干低级枯燥勾当就可
8safety_agent = Agent(name="Safety", model="fast-safety-check", ...)

奉立在殿堂上挂框让人当真言去膜拜不可违破之秘藏好规行矩方大总结 (Best practices)

  1. 一把钥匙配着专给那只孔去锁拿大不同的专属干活特列流工作干列: 跑外包分揽打发的分拣门房讲究图快(speed),居堂内给压断是非大定论的老道宗师重口口实讲足料厚(depth)
  2. 给干看大门兼设护城边卡关栏的看杂人用极其没几钱又快使型号供血就万事大吉: 防火阀门做安检就应该塞那种不值金银身价去开跑轻流的短道通就足够用
  3. 时时刻刻拿盯梢眼偷去窥并严把那这这帮子大干员来去往来过手大阵中连环派差时的转档推手黑箱记录: 动动眼角向侧门瞟一瞧 查数底账明细本档 (Request Logs) 这帮拉连台长走大剧到底吸骨敲髓费去多少大金币流水本
  4. 横眉给卡线强拧上最高跳跃周折来回路转回合数的发配紧箍咒轮轴最大数字设值: 免去由这等自己闭环绕轴干卷最后能造能造跑死牛不休不眠吃干库房的大无涯深渊回转环线死环里
  5. 铁腕死捆并勒实套模规约以结构化大体强绑拿结果出山交单的这事: 为那只要敢应并吐成堆数据包那必得全合乎全由您钦定大架构式去按套填槽

尚在此路边还能折回再拐向探路的门径指引余迹