内置官方工具：提供代码执行、文生图、在线搜索等实用工具。例如，Assistant 可以直接运行 Python 代码生成结果，调用搜索功能获取实时信息，或生成图片用于创意设计。 | # 示例代码，仅供参考
def submit_message(thread, assistant, message):
Messages.create(
thread_id=thread.id,
role="user",
content=message
)
run = Runs.create(
thread_id=thread.id,
assistant_id=assistant.id,
stream=True
)
for event, data in run:
if event == 'thread.message.delta':
yield data.delta.content.text.value
if event == 'thread.message.completed':
yield '\n'
if event == 'thread.run.step.delta':
# 第一次检测到step.delta时输出工具名称
if not hasattr(submit_message, 'tool_name_shown'):
submit_message.tool_name_shown = True
tool_name = data.delta.step_details.tool_calls[0]['type']
yield f"\n正在使用工具: {tool_name}\n\n"
formatted_output = format_tool_output(data.delta.step_details.tool_calls[0])
if formatted_output is not None:
yield formatted_output
|
内置对话管理：提供上下文管理工具，无需手动维护对话历史。 | # 示例代码，仅供参考
while True:
for event, data in run:
elif event == 'thread.run.requires_action':
# 工具调用 => 可能需要提交工具输出 => 导致新的 run 生成器
tool_calls = data.required_action.submit_tool_outputs.tool_calls
if not tool_calls:
continue
tool_outputs = []
for tool_call in tool_calls:
name = tool_call.function.name
arguments = json.loads(tool_call.function.arguments)
output = tools_map[name](**arguments)
# 普通工具
tool_outputs.append({"output": output})
# 将工具输出提交给 run，会返回一个新的 run 生成器
run = Runs.submit_tool_outputs(
thread_id=thread.id, # 原生的对话线程，无需手动管理
run_id=data.id,
tool_outputs=tool_outputs,
stream=True
)
yield "转接中...\n"
break # 跳出当前事件循环，用返回的 new_run 接着处理
elif event in ('thread.run.completed', 'thread.run.cancelled',
'thread.run.expired', 'thread.run.failed'):
# 当前 run 结束
break
else:
# for循环正常结束(没有 break)，说明 run 流耗尽
break
|
快速搭建多智能体系统：提供 Assistant、上下文、消息封装、流程控制等简易模板，可以灵活、高效地实现多智能体系统。例如具备自动规划能力的 Multi Agent 系统。 | # 示例代码，仅供参考
# 获得Multi Agent的回复，输入与输出需要与Gradio前端展示界面中的参数对齐
def get_multi_agent_response(query,history):
# 处理输入为空的情况
if len(query) == 0:
return "",history+[("","")],"",""
# 获取Agent的运行顺序
assistant_order = get_agent_response(PlannerAssistant,query)
try:
order_stk = ast.literal_eval(assistant_order)
cur_query = query
# 依次运行Agent
for i in range(len(order_stk)):
yield "----->".join(order_stk),history+[(query,"multi agent正在努力工作中...")],f"{order_stk[i]}正在处理信息...",""
cur_assistant = assistant_mapper[order_stk[i]]
response = get_agent_response(cur_assistant,cur_query)
yield "----->".join(order_stk),history+[(query,"multi agent正在努力工作中...")],response,""
# 如果当前Agent为最后一个Agent，则将其输出作为Multi Agent的输出
if i == len(order_stk)-1:
yield "----->".join(order_stk),history+[(query,response)],"assistant已处理完毕",""
# 如果当前Agent不是最后一个Agent，则将上一个Agent的输出response添加到下一轮的query中，作为参考信息
else:
# 在参考信息前后加上特殊标识符，可以防止大模型混淆参考信息与提问
cur_query = f"你可以参考已知的信息：\n{response}\n你要完整地回答用户的问题。问题是：{query}。"
# 兜底策略，如果上述程序运行失败，则直接调用ChatAssistant
except Exception as e:
yield "ChatAssistant",[(query,get_agent_response(ChatAssistant,query))],"",""
|
