当前位置： 首页 > news >正文

Agents-SDK智能体开发[2]之工具调用

news 2025/8/2 10:55:30

文章目录

  • 说明
  • 一 Agents SDK调用外部工具
    • 1.1 Agents SDK Function calling实现流程
    • 1.2 Agents SDK Runner loop循环执行
      • 1.2.1 Runner循环执行机制
      • 1.2.2 Runner循环执行验证
    • 1.3 Agents SDK接入更多自定义工具
      • 1.3.1 python_inter函数创建与使用
      • 1.3.2 sql_inter函数创建与使用
    • 1.4 Agents SDK多工具并联&串联执行
      • 1.4.1 Agents SDK多工具并联调用
      • 1.4.2 Agents SDK多工具串联调用

说明

  • 本文学自赋范社区公开课，仅供学习和交流使用，不用作任何商业用途！
  • OPENAI_API_KEY请到openai-hk.com获取！
  • OPENWEATHER_API_KEY请到openweathermap.org获取!

一 Agents SDK调用外部工具

  • Multi-Agent框架不仅能够顺利调用多个外部工具，还需要能够在多个不同的Agent中进行切换，以便于执行不同类型任务。
  • Agents SDK在进行外部工具调用的时候，并没有通过某些方法让模型诞生Function calling功能，而是通过一些流程，增加了Function calling执行的稳定性、以及简化了调用代码。例如，对于DeepSeek-V3模型来说，原始的Function calling执行流程如下：
    在这里插入图片描述
  • 查询当前天气时，让大模型调用外部工具的function calling的过程就如图：
    在这里插入图片描述
    在这里插入图片描述
  • Agents SDK调用外部工具的流程相对来说简单。简单测试外部天气查询工具
from agents import function_tool
import requests,json
OPENWEATHER_API_KEY="xxx"
def get_weather_test(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": OPENWEATHER_API_KEY,    # 输入自己的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)
res=get_weather_test(loc="Beijing") 
print(res)
# 先解析 JSON 字符串，再访问 weather 字段
weather_data = json.loads(res)["weather"]
print(weather_data)
  • 使用Python装饰器，增加一个get_weather函数
from openai import AsyncOpenAI
from agents import function_tool
import requests,json
from agents import OpenAIChatCompletionsModel,Agent,Runner,set_default_openai_client
OPENWEATHER_API_KEY="xxx"
OPENAI_API_KEY="hk-xxx"
OPENAI_API_BASE="https://api.openai-hk.com/v1"
MODEL="deepseek-v3"# 创建一个Agent对象并调用DeepSeek模型
external_client = AsyncOpenAI(base_url =OPENAI_API_BASE,api_key=OPENAI_API_KEY,
)set_default_openai_client(external_client)
deepseek_model=OpenAIChatCompletionsModel(model=MODEL,openai_client=external_client)
@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": OPENWEATHER_API_KEY,    # 输入自己的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)weather_agent = Agent(name="天气查询Agent",instructions="你是一名助人为乐的助手，并且可以进行天气信息查询",tools=[get_weather],model=deepseek_model
)  
weather_result = await Runner.run(weather_agent, input="你好，请问今天郑州天气如何？")
print(weather_result.final_output)
  • 今天郑州的天气晴朗，气温为36.08°C，体感温度较高，约为39.66°C。湿度为41%，风速为3.86米/秒，风向为42度。能见度非常好，达到10000米。总体来说，今天郑州天气炎热，请注意防暑降温！

1.1 Agents SDK Function calling实现流程

  • 在Agents SDK能够快速调用外部工具的背后，实际上是在高度自动化的工具下，完整运行了一次Function calling。具体完整的运行事件可以在weather_result中查看。发生的事件包括，一次外部函数请求事件（ToolCallltem）。
    len(weather_result.new_items) # 3

weather_result.new_items[0]

ToolCallItem(
agent=Agent(name='天气查询Agent', handoff_description=None, tools=[FunctionTool(name='get_weather', description='查询即时天气函数', params_json_schema={'properties': {'loc': {'description': "必要参数，字符串类型，用于表示查询天气的具体城市名称，    注意，中国的城市需要用对应城市的英文名称代替，例如如果需要查询北京市天气，则loc参数需要输入'Beijing'；", 'title': 'Loc'}}, 
'required': ['loc'], 'title': 'get_weather_args', 'type': 'object', 'additionalProperties': False}, on_invoke_tool=<function function_tool.<locals>._create_function_tool.<locals>._on_invoke_tool at 0x00000240E2426FC0>, strict_json_schema=True, is_enabled=True)], mcp_servers=[], mcp_config={}, 
instructions='你是一名助人为乐的助手，并且可以进行天气信息查询', prompt=None, handoffs=[], model=<agents.models.openai_chatcompletions.OpenAIChatCompletionsModel object at 0x00000240E1C724E0>, model_settings=ModelSettings(temperature=None, top_p=None, 
frequency_penalty=None, presence_penalty=None, tool_choice=None, parallel_tool_calls=None, truncation=None, max_tokens=None, reasoning=None, metadata=None, store=None, include_usage=None, response_include=None, extra_query=None, extra_body=None, extra_headers=None, extra_args=None), 
input_guardrails=[], output_guardrails=[], output_type=None, hooks=None, tool_use_behavior='run_llm_again', reset_tool_choice=True), 
raw_item=ResponseFunctionToolCall(arguments='{"loc":"Zhengzhou"}', call_id='1a1d38e364ad4ecf809c200fbc48ea4e', name='get_weather', type='function_call', id='__fake_id__', status=None), type='tool_call_item')
  • 在进行外部函数调用的时候，Agents SDK自动生成的tools参数列表，就相当于是Function calling的json schema对象，只不过是从我们定义的外部函数的函数说明中自动读取的。
    在这里插入图片描述
  • 需要在编写外部函数的时候谨慎的编写函数说明，否则会影响外部函数调用的准确率。
  • 外部函数请求过程就包括请求的外部函数名称以及相关函数等：
    在这里插入图片描述
  • 一个外部函数响应消息（ToolCallOutputItem）
weather_result.new_items[1]

ToolCallOutputItem(agent=Agent(name='天气查询Agent', handoff_description=None, tools=[FunctionTool(name='get_weather', description='查询即时天气函数', params_json_schema={'properties': {'loc': 
{'description': "必要参数，字符串类型，用于表示查询天气的具体城市名称，    注意，中国的城市需要用对应城市的英文名称代替，例如如果需要查询北京市天气，则loc参数需要输入'Beijing'；", 'title': 'Loc'}}, 'required': ['loc'], 'title': 'get_weather_args', 'type': 'object', 'additionalProperties': False}, on_invoke_tool=<function function_tool.<locals>._create_function_tool.<locals>._on_invoke_tool at 0x00000240E2426FC0>, strict_json_schema=True, is_enabled=True)], 
mcp_servers=[], mcp_config={}, instructions='你是一名助人为乐的助手，并且可以进行天气信息查询', prompt=None, handoffs=[], model=<agents.models.openai_chatcompletions.OpenAIChatCompletionsModel object at 0x00000240E1C724E0>, model_settings=ModelSettings(temperature=None, top_p=None, frequency_penalty=None, presence_penalty=None, tool_choice=None, parallel_tool_calls=None, truncation=None, max_tokens=None, reasoning=None, metadata=None, store=None, include_usage=None, response_include=None, extra_query=None, extra_body=None, extra_headers=None, extra_args=None), input_guardrails=[], output_guardrails=[], output_type=None, hooks=None, tool_use_behavior='run_llm_again', reset_tool_choice=True), 
raw_item={'call_id': '1a1d38e364ad4ecf809c200fbc48ea4e', 'output': '{"coord": {"lon": 113.6486, "lat": 34.7578}, 
"weather": [{"id": 800, "main": "Clear", "description": "\\u6674", "icon": "01d"}], "base": "stations", "main": {"temp": 36.27, "feels_like": 38.98, "temp_min": 36.27, "temp_max": 36.27, "pressure": 997, "humidity": 38, "sea_level": 997, "grnd_level": 983}, "visibility": 10000, "wind": {"speed": 4.26, "deg": 42, "gust": 5.01}, "clouds": {"all": 8}, "dt": 1754035113, 
"sys": {"country": "CN", "sunrise": 1753997701, "sunset": 1754047700}, "timezone": 28800, "id": 1784658, "name": "Zhengzhou", "cod": 200}', 'type': 'function_call_output'}, 
output='{"coord": {"lon": 113.6486, "lat": 34.7578}, "weather": [{"id": 800, "main": "Clear", "description": "\\u6674", "icon": "01d"}], "base": "stations", 
"main": {"temp": 36.27, "feels_like": 38.98, "temp_min": 36.27, "temp_max": 36.27, "pressure": 997, "humidity": 38, "sea_level": 997, "grnd_level": 983}, "visibility": 10000, "wind": {"speed": 4.26, "deg": 42, "gust": 5.01}, "clouds": {"all": 8}, "dt": 1754035113, 
"sys": {"country": "CN", "sunrise": 1753997701, "sunset": 1754047700}, "timezone": 28800, "id": 1784658, "name": "Zhengzhou", "cod": 200}', type='tool_call_output_item')
  • 既外部函数运行结果，也就是查询到的天气结果：
{'call_id': '1a1d38e364ad4ecf809c200fbc48ea4e','output': '{"coord": {"lon": 113.6486,"lat": 34.7578},"weather": [{"id": 800,"main": "Clear","description": "\\u6674","icon": "01d"}],"base": "stations","main": {"temp": 36.27,"feels_like": 38.98,"temp_min": 36.27,"temp_max": 36.27,"pressure": 997,"humidity": 38,"sea_level": 997,"grnd_level": 983},"visibility": 10000,"wind": {"speed": 4.26,"deg": 42,"gust": 5.01},"clouds": {"all": 8},"dt": 1754035113,"sys": {"country": "CN","sunrise": 1753997701,"sunset": 1754047700},"timezone": 28800,"id": 1784658,"name": "Zhengzhou","cod": 200}','type': 'function_call_output'
}
  • 还包含一条最终模型响应事件（MessageOutputItem）
{MessageOutputItem(agent=Agent(name='天气查询Agent',handoff_description=None,tools=[FunctionTool(name='get_weather',description='查询即时天气函数',params_json_schema={'properties': {'loc': {'description': "必要参数，字符串类型，用于表示查询天气的具体城市名称，    注意，中国的城市需要用对应城市的英文名称代替，例如如果需要查询北京市天气，则loc参数需要输入'Beijing'；",'title': 'Loc'}},'required': ['loc'],'title': 'get_weather_args','type': 'object','additionalProperties': False},on_invoke_tool=<functionfunction_tool.<locals>._create_function_tool.<locals>._on_invoke_toolat0x00000240E2426FC0>,strict_json_schema=True,is_enabled=True)],mcp_servers=[],mcp_config={},instructions='你是一名助人为乐的助手，并且可以进行天气信息查询',prompt=None,handoffs=[],model=<agents.models.openai_chatcompletions.OpenAIChatCompletionsModelobjectat0x00000240E1C724E0>,model_settings=ModelSettings(temperature=None,top_p=None,frequency_penalty=None,presence_penalty=None,tool_choice=None,parallel_tool_calls=None,truncation=None,max_tokens=None,reasoning=None,metadata=None,store=None,include_usage=None,response_include=None,extra_query=None,extra_body=None,extra_headers=None,extra_args=None),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='今天郑州的天气晴朗，气温为36.27°C，体感温度较高，约38.98°C。湿度为38%，风速4.26米/秒，风向42度。能见度良好，达到10000米。总体来说，天气炎热，请注意防暑降温！',type='output_text',logprobs=None)],role='assistant',status='completed',type='message'),type='message_output_item')
}
  • 模型最终回复：
{ResponseOutputMessage(id='__fake_id__',content=[ResponseOutputText(annotations=[],text='今天郑州的天气晴朗，气温为36.27°C，体感温度较高，约38.98°C。湿度为38%，风速4.26米/秒，风向42度。能见度良好，达到10000米。总体来说，天气炎热，请注意防暑降温！',type='output_text',logprobs=None)],role='assistant',status='completed',type='message')
}
  • 总结：在调用Agents SDK时，我们只需要在tools参数位上输入对应的函数，就能自动执行Function calling流程。
    在这里插入图片描述

1.2 Agents SDK Runner loop循环执行

1.2.1 Runner循环执行机制

  • Agents SDK能够高效快速执行Function calling，得益于Runner的循环执行机制。
  • Runner的循环执行机制从指定的起始 Agent 开始运行一个工作流。Agent 会在一个循环中运行，直到生成最终输出为止。
    在这里插入图片描述

循环流程如下：

  1. 使用提供的输入调用 Agent。
  2. 如果产生了“最终输出”（例如返回了 agent.output_type 类型的内容），循环终止。
  3. 如果产生了 handoff（交接），则使用新的 Agent 重新进入循环。
  4. 否则，会执行工具调用（tool calls，如有），然后重新进入循环。

在以下两种情况下，Agent 会抛出异常：

  1. 如果超过了最大轮数 max_turns，则会抛出 MaxTurnsExceeded 异常。
  2. 如果触发了 Guardrail 的“绊线”机制（tripwire），则会抛出 GuardrailTripwireTriggered 异常。
    ⚠️ 注意：只有第一个 Agent 的输入会被检查 Guardrails（安全检查机制）。
Runner.run(starting_agent: 'Agent[TContext]',input: 'str | list[TResponseInputItem]',*,context: 'TContext | None' = None,max_turns: 'int' = 10,hooks: 'RunHooks[TContext] | None' = None,run_config: 'RunConfig | None' = None,
) -> 'RunResult'

📌 参数说明：

参数名类型说明
starting_agentAgent[TContext]运行起点的 Agent（智能体）
inputstr 或 list[TResponseInputItem]Agent 的初始输入。你可以传入一个字符串（如用户消息），也可以传入一个输入项的列表。
contextTContext 或 None要与 Agent 一起运行的上下文（比如用户资料、状态等）
max_turnsintAgent 最多运行的轮数（默认是 10）。每次 AI 响应（包括工具调用）都算一轮。
hooksRunHooks[TContext] 或 None生命周期钩子对象。你可以通过它注册回调函数，监听 Agent 的各种事件（如每轮开始、每次 tool 调用等）。
run_configRunConfig 或 None对整个 Agent 运行过程的全局配置选项。

1.2.2 Runner循环执行验证

from openai import AsyncOpenAI
from agents import function_tool
import requests,json
from agents import OpenAIChatCompletionsModel,Agent,Runner,set_default_openai_clientOPENWEATHER_API_KEY="xxx"
OPENAI_API_KEY="hk-xxx"
OPENAI_API_BASE="https://api.openai-hk.com/v1"
MODEL="deepseek-v3"# 创建一个Agent对象并调用DeepSeek模型
external_client = AsyncOpenAI(base_url =OPENAI_API_BASE,api_key=OPENAI_API_KEY,
)set_default_openai_client(external_client)
deepseek_model=OpenAIChatCompletionsModel(model=MODEL,openai_client=external_client)@function_tool
def test_write_file(content):"""将指定内容写入本地文件。:param content: 必要参数，字符串类型，用于表示需要写入文档的具体内容。:return：是否成功写入"""raise RuntimeError("由于格式问题，无法写入本地文档，请调整格式后重新尝试。")test_agent = Agent(name="写入本地文件Agent",instructions="你是一名助人为乐的助手，并且将部分文档写入本地",tools=[test_write_file],model=deepseek_model
)
test_result = await Runner.run(test_agent, input="请将《DeepSeek智能体开发实战》标题写入本地文档")
print(test_result.final_output)
  • 执行结果

很抱歉多次尝试后仍然无法成功写入文档。可能是由于以下原因之一：

  1. 系统对写入内容有严格的格式限制（如必须包含特定前缀/后缀）
  2. 当前工具函数存在配置问题
  3. 文件写入权限限制

建议您：

  1. 检查文档写入格式要求
  2. 提供更具体的写入格式示例
  3. 或尝试其他方式保存内容

如果您能提供具体的格式要求，我很乐意再次尝试。

  • 模型会尝试多次：

    print(len(test_result.new_items))
print(test_result.new_items[0].raw_item)
print("--")
print(test_result.new_items[1].raw_item)
print("--")
print(test_result.new_items[2].raw_item)
print("--")
print(test_result.new_items[3].raw_item)
print("--")
print(test_result.new_items[4].raw_item)
print("--")
print(test_result.new_items[5].raw_item)
print("--")
print(test_result.new_items[6].raw_item)
print("--")
print(test_result.new_items[7].raw_item)
print("--")
print(test_result.new_items[8].raw_item)

    15
ResponseFunctionToolCall(arguments='{"content":"DeepSeek智能体开发实战"}', call_id='304149a2dcfd4ca6a6e174fd3ece1bea', name='test_write_file', type='function_call', id='__fake_id__', status=None)
--
{'call_id': '304149a2dcfd4ca6a6e174fd3ece1bea', 'output': 'An error occurred while running the tool. Please try again. Error: 由于格式问题，无法写入本地文档，请调整格式后重新尝试。', 'type': 'function_call_output'}
--
ResponseOutputMessage(id='__fake_id__', content=[ResponseOutputText(annotations=[], text='看来写入本地文档时遇到了格式问题，我会尝试调整内容格式后重新写入。请稍等。', type='output_text', logprobs=None)], role='assistant', status='completed', type='message')
--
ResponseFunctionToolCall(arguments='{"content":"# DeepSeek智能体开发实战"}', call_id='655ce77a54764ff68837a93d5732b678', name='test_write_file', type='function_call', id='__fake_id__', status=None)
--
{'call_id': '655ce77a54764ff68837a93d5732b678', 'output': 'An error occurred while running the tool. Please try again. Error: 由于格式问题，无法写入本地文档，请调整格式后重新尝试。', 'type': 'function_call_output'}
--
ResponseOutputMessage(id='__fake_id__', content=[ResponseOutputText(annotations=[], text='看来需要更简单的纯文本格式来写入文档。我再试一次：', type='output_text', logprobs=None)], role='assistant', status='completed', type='message')
--
ResponseFunctionToolCall(arguments='{"content":"DeepSeek智能体开发实战"}', call_id='b6b7f97c2e8a4e879fcd2804e27971f1', name='test_write_file', type='function_call', id='__fake_id__', status=None)
--
{'call_id': 'b6b7f97c2e8a4e879fcd2804e27971f1', 'output': 'An error occurred while running the tool. Please try again. Error: 由于格式问题，无法写入本地文档，请调整格式后重新尝试。', 'type': 'function_call_output'}
--
ResponseOutputMessage(id='__fake_id__', content=[ResponseOutputText(annotations=[], text='看来系统对写入内容有严格的格式要求。我再尝试以下两种方案：\n\n方案1：完全去除任何特殊符号', type='output_text', logprobs=None)], role='assistant', status='completed', type='message')
  • DeepSeek-V3-0324模型有自带的避免Function calling出现死循环的功能，最多调用若干次无法返回正确结果后，将会停止继续调用外部函数。

1.3 Agents SDK接入更多自定义工具

1.3.1 python_inter函数创建与使用

from openai import AsyncOpenAI
from agents import function_tool
import requests,json
from agents import OpenAIChatCompletionsModel,Agent,Runner,set_default_openai_clientOPENAI_API_KEY="hk-xxx"
OPENAI_API_BASE="https://api.openai-hk.com/v1"
MODEL="deepseek-v3"# 创建一个Agent对象并调用DeepSeek模型
external_client = AsyncOpenAI(base_url =OPENAI_API_BASE,api_key=OPENAI_API_KEY,
)set_default_openai_client(external_client)
deepseek_model=OpenAIChatCompletionsModel(model=MODEL,openai_client=external_client)# 使用装饰器标记这是一个函数工具，可能用于某些框架或系统的特殊处理
@function_tool
def python_inter(py_code):"""运行用户提供的 Python 代码，并返回执行结果。:param py_code: 字符串形式的 Python 代码:return: 代码运行的最终结果（JSON字符串形式）"""# 获取当前全局变量字典的引用g = globals()try:# 首先尝试将代码作为表达式求值# eval()用于执行单个Python表达式并返回结果result = eval(py_code, g)# 将结果转换为JSON字符串返回，ensure_ascii=False允许非ASCII字符return json.dumps(str(result), ensure_ascii=False)except Exception:# 如果不是表达式或eval失败，记录执行前的全局变量集合global_vars_before = set(g.keys())try:# 尝试用exec执行代码（适用于语句、函数定义等多行代码）exec(py_code, g)except Exception as e:# 如果执行出错，返回错误信息return json.dumps(f"代码执行时报错: {e}", ensure_ascii=False)# 获取执行后的全局变量集合global_vars_after = set(g.keys())# 计算新增加的全局变量new_vars = global_vars_after - global_vars_beforeif new_vars:# 如果有新变量，收集这些变量的值返回safe_result = {}for var in new_vars:try:# 尝试序列化变量值，确保可以转换为JSONjson.dumps(g[var])# 如果可以序列化，直接存储值safe_result[var] = g[var]except (TypeError, OverflowError):# 如果不能序列化（如自定义对象），则存储其字符串表示safe_result[var] = str(g[var])# 返回包含新变量的字典return json.dumps(safe_result, ensure_ascii=False)else:# 如果没有新变量产生，返回执行成功信息return json.dumps("已经顺利执行代码", ensure_ascii=False)python_agent = Agent(name="python_agent",instructions="你是一名助人为乐的助手，并且能调用本地Python环境进行编程",tools=[python_inter],model=deepseek_model
)python_result = await Runner.run(python_agent, input="请帮我用Python代码模拟一组数据，来绘制核密度分布图。注意添加代码设置图中中文内容正常显示")
print(python_result.final_output)

在这里插入图片描述
太好了！以下是代码生成的核密度分布图的关键信息总结：

  1. 图表内容

    • 蓝色填充的核密度曲线，平滑展示了数据的概率分布
    • 网格线（灰色虚线）辅助观察数值分布
    • 标题为"核密度分布图示例"，X/Y轴分别标注"数值"和"密度"

  2. 中文显示

    • 已通过plt.rcParams配置黑体(SimHei)显示中文
    • 解决了坐标轴负号显示问题

  3. 技术细节

    • 使用seaborn.kdeplot自动计算最佳带宽
    • fill=True参数替代了旧版shade=True的填充方式
    • 图形尺寸为10x6英寸

如果需要调整任何细节（如颜色、透明度、添加数据统计信息等），可以随时告诉我！

1.3.2 sql_inter函数创建与使用

  • 安装pymysql库
pip install pymysql
  • 完整的工具创建与使用代码
from openai import AsyncOpenAI
from agents import function_tool
import requests,json
from agents import OpenAIChatCompletionsModel,Agent,Runner,set_default_openai_clientOPENAI_API_KEY="hk-xxx" # 请到openai-hk.com自行获取api_key
OPENAI_API_BASE="https://api.openai-hk.com/v1"
MODEL="deepseek-v3"# 创建一个Agent对象并调用DeepSeek模型
external_client = AsyncOpenAI(base_url =OPENAI_API_BASE,api_key=OPENAI_API_KEY,
)set_default_openai_client(external_client)
deepseek_model=OpenAIChatCompletionsModel(model=MODEL,openai_client=external_client)# 导入pymysql模块，用于连接和操作MySQL数据库
import pymysql# 使用装饰器标记这个函数是一个工具函数（具体装饰器功能取决于function_tool的实现）
@function_tool
def sql_inter(sql_query):"""查询本地MySQL数据库，通过运行一段SQL代码来进行数据库查询。参数:sql_query (str): 字符串形式的SQL查询语句，用于执行对MySQL中school数据库中各张表进行查询，并获得各表中的各类相关信息。注意：虽然文档说查询school数据库，但实际代码中连接的是telco_db数据库。返回:str: 将SQL查询结果转换为JSON格式字符串返回。如果查询没有结果，则返回空数组的JSON表示。注意:1. 这是一个敏感操作，直接执行传入的SQL语句，存在SQL注入风险2. 连接使用固定的数据库凭证，在生产环境中应该使用更安全的方式管理凭证3. 连接使用utf8字符集，确保能正确处理中文等非ASCII字符4. 无论查询是否成功，都会确保数据库连接被关闭"""# 建立数据库连接connection = pymysql.connect(host='localhost',  # 数据库服务器地址，这里是本地port=3306,user='root',      # 数据库用户名，通常具有最高权限passwd='yqk.20021027',     # 数据库密码，示例中使用的是简单密码，实际环境应使用强密码db='data_agent',    # 要连接的数据库名称（注意与文档描述中的school数据库不一致）charset='utf8'    # 设置字符集为utf8，支持多语言字符)try:# 创建游标对象，用于执行SQL语句with connection.cursor() as cursor:# 直接使用传入的SQL查询语句sql = sql_query# 执行SQL查询cursor.execute(sql)# 获取所有查询结果results = cursor.fetchall()  # 返回的是元组格式的结果集finally:# 无论是否发生异常，都确保关闭数据库连接connection.close()# 将查询结果转换为JSON格式字符串并返回# 注意：需要确保results中的数据类型可以被json.dumps序列化return json.dumps(results)SQL_agent = Agent(name="sql_agent",instructions="你是一名助人为乐的助手，并且能连接本地MySQL数据库进行数据查询",tools=[sql_inter],model=deepseek_model
)sql_result = await Runner.run(SQL_agent, input="请帮我查询当前数据库中，总共有几张表。")print(sql_result.final_output)

当前数据库中总共有5张表，分别是：1. classes
2. courses
3. schools
4. student_courses
5. students

1.4 Agents SDK多工具并联&串联执行

  • 对于Agents SDK来说，也同样支持Function calling的多种相应模式。
    在这里插入图片描述

1.4.1 Agents SDK多工具并联调用

在这里插入图片描述

  • 采用weather_agent来进行多地天气查询，即可测试Agents SDK是否会开启多工具并联调用。
    在这里插入图片描述
from openai import AsyncOpenAI
from agents import function_tool
import requests,json
from agents import OpenAIChatCompletionsModel,Agent,Runner,set_default_openai_clientOPENWEATHER_API_KEY="055bcd72c5dd30682f75465070ae0f46"
OPENAI_API_KEY="hk-xxx" # openai-hk.com
OPENAI_API_BASE="https://api.openai-hk.com/v1"
MODEL="deepseek-v3"# 创建一个Agent对象并调用DeepSeek模型
external_client = AsyncOpenAI(base_url =OPENAI_API_BASE,api_key=OPENAI_API_KEY,
)set_default_openai_client(external_client)
deepseek_model=OpenAIChatCompletionsModel(model=MODEL,openai_client=external_client)
@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": OPENWEATHER_API_KEY,    # 输入自己的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)weather_agent = Agent(name="天气查询Agent",instructions="你是一名助人为乐的助手，并且可以进行天气信息查询",tools=[get_weather],model=deepseek_model
)  
multi_weather_result = await Runner.run(weather_agent, input="你好，请问今天郑州和南阳天气如何？")
print("结果输出")
print(multi_weather_result.final_output)
print(f"本次响应总共的事件个数：{len(multi_weather_result.new_items)}")
print("首先是一次性发起同一个外部函数的两次调用请求")
print(multi_weather_result.new_items[0].raw_item)
print(multi_weather_result.new_items[1].raw_item)
print("然后获得了两个外部函数响应")
print(multi_weather_result.new_items[2].raw_item)
print(multi_weather_result.new_items[3].raw_item)
print("最终模型回复结果")
print(multi_weather_result.new_items[4].raw_item)
  • 运行结果
结果输出
今天郑州和南阳的天气情况如下：- **郑州**：晴，少云，气温36.6°C，体感温度41.2°C，湿度42%，风速4.7 m/s。
- **南阳**：多云，气温35.9°C，体感温度39.3°C，湿度41%，风速3.7 m/s。请注意防暑降温！本次响应总共的事件个数：5首先是一次性发起同一个外部函数的两次调用请求
ResponseFunctionToolCall(arguments='{"loc": "Zhengzhou"}', call_id='c983492150db41b69694af5dbc9a75ac', name='get_weather', type='function_call', id='__fake_id__', status=None)
ResponseFunctionToolCall(arguments='{"loc": "Nanyang"}', call_id='01752858d0a9453b880137272e834a35', name='get_weather', type='function_call', id='__fake_id__', status=None)然后获得了两个外部函数响应
{'call_id': 'c983492150db41b69694af5dbc9a75ac', 'output': '{"coord": {"lon": 113.6486, "lat": 34.7578}, 
"weather": [{"id": 801, "main": "Clouds", "description": "\\u6674\\uff0c\\u5c11\\u4e91", "icon": "02d"}], "base": "stations", 
"main": {"temp": 36.64, "feels_like": 41.24, "temp_min": 36.64, "temp_max": 36.64, "pressure": 998, "humidity": 42, "sea_level": 998, "grnd_level": 983}, "visibility": 10000, "wind": {"speed": 4.7, "deg": 48, "gust": 5.12}, "clouds": {"all": 13}, "dt": 1754039700, 
"sys": {"country": "CN", "sunrise": 1753997701, "sunset": 1754047700}, "timezone": 28800, "id": 1784658, "name": "Zhengzhou", "cod": 200}', 'type': 'function_call_output'}{'call_id': '01752858d0a9453b880137272e834a35', 'output': '{"coord": {"lon": 112.5328, "lat": 32.9947}, "weather": [{"id": 802, "main": "Clouds", "description": "\\u591a\\u4e91", "icon": "03d"}], "base": "stations", 
"main": {"temp": 35.91, "feels_like": 39.31, "temp_min": 35.91, "temp_max": 35.91, "pressure": 997, "humidity": 41, "sea_level": 997, "grnd_level": 982}, "visibility": 10000, "wind": {"speed": 3.73, "deg": 64, "gust": 5.17}, "clouds": {"all": 27}, "dt": 1754039727, 
"sys": {"country": "CN", "sunrise": 1753998181, "sunset": 1754047756}, "timezone": 28800, "id": 1799629, "name": "Nanyang", "cod": 200}', 'type': 'function_call_output'}最终模型回复结果
ResponseOutputMessage(id='__fake_id__', 
content=[ResponseOutputText(annotations=[], 
text='今天郑州和南阳的天气情况如下：\n\n
- **郑州**：晴，少云，气温36.6°C，体感温度41.2°C，湿度42%，风速4.7 m/s。\n
- **南阳**：多云，气温35.9°C，体感温度39.3°C，湿度41%，风速3.7 m/s。\n\n请注意防暑降温！', 
- type='output_text', logprobs=None)], role='assistant', status='completed', type='message')

1.4.2 Agents SDK多工具串联调用

在这里插入图片描述

  • 采用weather_agent来进行多地天气查询，然后使用write_file函数，用于将“文本写入本地”，即可测试Agents SDK是否会开启多工具串联调用。
    在这里插入图片描述
from openai import AsyncOpenAI
from agents import function_tool
import requests,json
from agents import OpenAIChatCompletionsModel,Agent,Runner,set_default_openai_clientOPENWEATHER_API_KEY="xxx"
OPENAI_API_KEY="hk-xxx"
OPENAI_API_BASE="https://api.openai-hk.com/v1"
MODEL="deepseek-v3"# 创建一个Agent对象并调用DeepSeek模型
external_client = AsyncOpenAI(base_url =OPENAI_API_BASE,api_key=OPENAI_API_KEY,
)set_default_openai_client(external_client)
deepseek_model=OpenAIChatCompletionsModel(model=MODEL,openai_client=external_client)
@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": OPENWEATHER_API_KEY,    # 输入自己的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)@function_tool
def write_file(content):"""将指定内容写入本地文件。:param content: 必要参数，字符串类型，用于表示需要写入文档的具体内容。:return：字符串，表示是否成功写入"""try:with open("res.md", "w", encoding="utf-8") as file:file.write(content)return "已成功写入本地文件(res.md)"except Exception as e:return f"写入文件失败: {str(e)}"weather_agent = Agent(name="天气查询Agent",instructions="你是一名助人为乐的助手，并且可以进行天气信息查询",tools=[get_weather, write_file],model=deepseek_model
)  
result = await Runner.run(weather_agent, input="请查询今天郑州和南阳天气，并写入本地文件中。")
print("结果输出")
print(result.final_output)
print(f"本次响应总共的事件个数：{len(multi_weather_result.new_items)}")
print("首先查询天气的外部函数的两次并行调用请求")
print(result.new_items[0].raw_item)
print(result.new_items[1].raw_item)
print("然后获得了两个外部函数响应")
print(result.new_items[2].raw_item)
print(result.new_items[3].raw_item)
print("发起写入本地文件的外部函数的两次并行调用请求")
print(result.new_items[4].raw_item)
print(result.new_items[5].raw_item)
print("模型回答用户问题")
print(result.new_items[5].raw_item)
print("消息列表如下：")
print(result.to_input_list())

结果输出
已将郑州和南阳的天气信息成功写入本地文件 `res.md` 中。如果需要进一步帮助，请随时告诉我！
本次响应总共的事件个数：5
首先查询天气的外部函数的两次并行调用请求
ResponseFunctionToolCall(arguments='{"loc": "Zhengzhou"}', call_id='fedc9fa70f924ff9bec1c0900db7b7b1', name='get_weather', type='function_call', id='__fake_id__', status=None)
ResponseFunctionToolCall(arguments='{"loc": "Nanyang"}', call_id='8382ebe915be429aa4f63787a1968a6d', name='get_weather', type='function_call', id='__fake_id__', status=None)
然后获得了两个外部函数响应
{'call_id': 'fedc9fa70f924ff9bec1c0900db7b7b1', 'output': '{"coord": {"lon": 113.6486, "lat": 34.7578}, "weather": xxx}
{'call_id': '8382ebe915be429aa4f63787a1968a6d', 'output': '{"coord": {"lon": 112.5328, "lat": 32.9947}, "weather": xxx}
发起写入本地文件的外部函数的两次并行调用请求
ResponseFunctionToolCall(arguments='{"content":"郑州天气：晴，少云，温度35.01°C，体感温度34.69°C，湿度30%，风速3.41m/s。\\n南阳天气：晴，少云，温度34.61°C，体感温度39.4°C，湿度49%，风速3.03m/s。"}', call_id='68bd8ce2254e4e57847d4fa03a40e749', name='write_file', type='function_call', id='__fake_id__', status=None)
{'call_id': '68bd8ce2254e4e57847d4fa03a40e749', 'output': '已成功写入本地文件(res.md)', 'type': 'function_call_output'}
模型回答用户问题
{'call_id': '68bd8ce2254e4e57847d4fa03a40e749', 'output': '已成功写入本地文件(res.md)', 'type': 'function_call_output'}
消息列表如下：
[{'content': '请查询今天郑州和南阳天气，并写入本地文件中。','role': 'user'
},
{'arguments': '{"loc": "Zhengzhou"}','call_id': 'fedc9fa70f924ff9bec1c0900db7b7b1','name': 'get_weather','type': 'function_call','id': '__fake_id__'
},
{'arguments': '{"loc": "Nanyang"}','call_id': '8382ebe915be429aa4f63787a1968a6d','name': 'get_weather','type': 'function_call','id': '__fake_id__'
},
{'call_id': 'fedc9fa70f924ff9bec1c0900db7b7b1','output': '{"coord": {"lon": 113.6486,"lat": 34.7578},"weather": [{"id": 801,"main": "Clouds","description": "\\u6674\\uff0c\\u5c11\\u4e91","icon": "02d"}],"base": "stations","main": {"temp": 35.01,"feels_like": 34.69,"temp_min": 35.01,"temp_max": 35.01,"pressure": 996,"humidity": 30,"sea_level": 996,"grnd_level": 983},"visibility": 10000,"wind": {"speed": 3.41,"deg": 51,"gust": 5.04},"clouds": {"all": 15},"dt": 1754040638,"sys": {"country": "CN","sunrise": 1753997701,"sunset": 1754047700},"timezone": 28800,"id": 1784658,"name": "Zhengzhou","cod": 200}','type': 'function_call_output'
},
{'call_id': '8382ebe915be429aa4f63787a1968a6d','output': '{"coord": {"lon": 112.5328,"lat": 32.9947},"weather": [{"id": 801,"main": "Clouds","description": "\\u6674\\uff0c\\u5c11\\u4e91","icon": "02d"}],"base": "stations","main": {"temp": 34.61,"feels_like": 39.4,"temp_min": 34.61,"temp_max": 34.61,"pressure": 997,"humidity": 49,"sea_level": 997,"grnd_level": 982},"visibility": 10000,"wind": {"speed": 3.03,"deg": 86,"gust": 4.4},"clouds": {"all": 24},"dt": 1754040736,"sys": {"country": "CN","sunrise": 1753998181,"sunset": 1754047756},"timezone": 28800,"id": 1799629,"name": "Nanyang","cod": 200}','type': 'function_call_output'
},
{'arguments': '{"content": "郑州天气：晴，少云，温度35.01°C，体感温度34.69°C，湿度30%，风速3.41m/s。\\n南阳天气：晴，少云，温度34.61°C，体感温度39.4°C，湿度49%，风速3.03m/s。"}','call_id': '68bd8ce2254e4e57847d4fa03a40e749','name': 'write_file','type': 'function_call','id': '__fake_id__'
},
{'call_id': '68bd8ce2254e4e57847d4fa03a40e749','output': '已成功写入本地文件(res.md)','type': 'function_call_output'
},
{'id': '__fake_id__','content': [{'annotations': [],'text': '已将郑州和南阳的天气信息成功写入本地文件`res.md`中。如果需要进一步帮助，请随时告诉我！','type': 'output_text'}],'role': 'assistant','status': 'completed','type': 'message'
}]
http://www.dtcms.com/a/310430.html

相关文章：

  • Nginx 来正确地托管网站服务
  • 《软件测试与质量控制》实验报告一 测试用例设计
  • 自动化框架pytest
  • 小学阶段的学习机推荐：科大讯飞T30、Lumie 10学习机暑期16项AI功能升级
  • 2025电赛G题-发挥部分-参数自适应FIR滤波器
  • python列表推导式
  • uniapp基础 （二）
  • 电商作图，商品图、模特图、促销海报设计
  • Unity优化技巧：自动隐藏视野外的3D模型
  • 【人工智能-16】机器学习：概念、工具介绍、数据集、特征工程
  • 铁皮矫平机进阶小百科
  • C# _Json数据
  • MySQL 45 讲 18-20
  • React 19 革命性升级：编译器自动优化，告别手动性能调优时代
  • 携程PMO资深经理、携程技术委员会人工智能委员会秘书陈强受邀为PMO大会主持人
  • 开源vGPU解决方案HAMi
  • 2025.8.1
  • python中appium 的NoSuchElementException错误 原因以及解决办法
  • C++基础语法
  • Redis实战（5）-- 高级数据结构 HyperLogLog
  • 调整Idea缓存目录，释放C盘空间
  • UniApp与WebView双向通信机制及生产级实现方案全解析
  • 振动波形转音频播放并做声纹聚类
  • 【数据分享】南海综合波浪数据（1945-2018 年）（获取方式看文末）
  • 【历史人物】【王安石】简历与生平
  • win11怎么看本机ip地址？怎么查看代理端口？
  • SAP Datasphere 03 - 数据权限
  • OpenShift AI - 用 Hardware profiles 为运行环境分配可用的硬件规格
  • 什么是股指期货的不对冲策略？
  • 【Flutter】内存泄漏总结