通过Python脚本示例快速上手Taotoken的流式响应与函数调用

🚀 告别海外账号与网络限制！稳定直连全球优质大模型，限时半价接入中。 👉 点击领取海量免费额度

通过Python脚本示例快速上手Taotoken的流式响应与函数调用

在熟悉了如何通过Taotoken平台进行基础的模型调用后，开发者往往会希望利用更高级的API特性来构建体验更佳、能力更强的应用。流式响应能够实现类似打字机效果的实时内容输出，提升交互感；而函数调用功能则让大模型具备了与外部工具或代码逻辑交互的能力。本文将基于Python环境，演示如何通过Taotoken的OpenAI兼容API，快速实现这两项进阶功能。

1. 准备工作与环境配置

在开始编写流式响应或函数调用的代码之前，请确保你已经完成了最基础的接入准备。这包括在Taotoken控制台创建了API Key，并在模型广场确认了你想要调用的模型ID。本文的示例将使用claude-sonnet-4-6模型，你可以根据实际需求替换为其他支持的模型。

确保你已安装了最新版本的openaiPython SDK。Taotoken平台完全兼容OpenAI SDK的调用方式，你只需要在初始化客户端时指定正确的base_url。

from openai import OpenAI # 初始化客户端，指向Taotoken的OpenAI兼容端点 client = OpenAI( api_key="你的Taotoken_API_Key", # 请替换为实际的API Key base_url="https://taotoken.net/api", # 关键：使用此Base URL )

请将代码中的你的Taotoken_API_Key替换为你在控制台获取的真实密钥。这个初始化步骤是后续所有调用的基础。

2. 实现聊天补全的流式响应

流式响应（Streaming Response）允许服务器在生成完整回复的过程中，就将已生成的部分内容分块（chunk）发送给客户端。这对于需要实时显示模型思考过程或构建聊天界面的应用至关重要。

在OpenAI SDK中，只需在调用chat.completions.create方法时，将stream参数设置为True即可开启流式响应。随后，你需要遍历返回的生成器对象来处理每一个数据块。

def stream_chat_completion(): stream = client.chat.completions.create( model="claude-sonnet-4-6", messages=[ {"role": "system", "content": "你是一个乐于助人的助手。"}, {"role": "user", "content": "请用一段话介绍流式响应的优点。"} ], stream=True, # 启用流式输出 max_tokens=500, ) collected_content = [] print("模型回复（流式）: ", end="", flush=True) for chunk in stream: # 每个chunk是一个ChatCompletionChunk对象 if chunk.choices and chunk.choices[0].delta.content is not None: content_delta = chunk.choices[0].delta.content print(content_delta, end="", flush=True) # 逐块打印，模拟打字效果 collected_content.append(content_delta) full_response = "".join(collected_content) print(f"\n\n完整回复已接收，共{len(full_response)}个字符。") return full_response # 调用函数 if __name__ == "__main__": response = stream_chat_completion()

这段代码会逐字打印出模型的回复。chunk.choices[0].delta.content包含了当前数据块中新增加的文本内容。需要注意的是，流式响应中，finish_reason等字段可能会在最后一个数据块中返回。在实际应用中，你可以将每次收到的content_delta实时推送到前端页面或客户端，从而实现流畅的对话体验。

3. 利用函数调用扩展模型能力

函数调用（Function Calling）允许你向模型描述一组可用的工具（函数），模型在理解用户请求后，可以决定是否需要调用某个函数，并输出一个符合函数参数要求的结构化JSON。你的程序接收到这个调用请求后，实际执行函数（例如查询数据库、调用外部API、运行计算等），并将结果返回给模型，由模型整合信息后生成最终回答给用户。

下面是一个完整的示例，演示了如何定义函数、让模型决定是否调用、以及如何处理整个交互流程。

import json # 1. 定义可供模型调用的函数工具列表 tools = [ { "type": "function", "function": { "name": "get_current_weather", "description": "获取指定城市的当前天气情况", "parameters": { "type": "object", "properties": { "location": { "type": "string", "description": "城市名称，例如：北京、上海", }, "unit": { "type": "string", "enum": ["celsius", "fahrenheit"], "description": "温度单位，摄氏度或华氏度", } }, "required": ["location"], }, }, } ] # 2. 模拟一个实际执行天气查询的函数 def execute_function_call(tool_call): """ 根据模型请求的工具调用信息，执行本地函数并返回结果。 """ function_name = tool_call.function.name arguments = json.loads(tool_call.function.arguments) if function_name == "get_current_weather": location = arguments.get("location", "未知地点") unit = arguments.get("unit", "celsius") # 这里模拟一个固定的天气查询结果。真实场景中，你会在这里调用天气API。 weather_info = { "location": location, "temperature": 22 if unit == "celsius" else 72, "unit": unit, "forecast": ["晴朗", "微风"], "humidity": 65 } return json.dumps(weather_info, ensure_ascii=False) else: return json.dumps({"error": f"未知函数: {function_name}"}) # 3. 主对话函数，处理可能发生的函数调用循环 def chat_with_function_calling(user_query): messages = [{"role": "user", "content": user_query}] # 第一次调用，模型可能会请求调用函数 response = client.chat.completions.create( model="claude-sonnet-4-6", messages=messages, tools=tools, tool_choice="auto", # 由模型自动决定是否调用工具 ) response_message = response.choices[0].message messages.append(response_message) # 将模型的回复（可能包含工具调用请求）加入历史 # 4. 检查回复中是否包含工具调用请求 tool_calls = response_message.tool_calls if tool_calls: print(f"模型请求调用 {len(tool_calls)} 个工具。") # 遍历所有被请求的工具调用（可能同时有多个） for tool_call in tool_calls: function_result = execute_function_call(tool_call) print(f"执行函数 `{tool_call.function.name}`， 参数: {tool_call.function.arguments}") print(f"得到结果: {function_result}") # 将函数执行结果作为一条新消息追加到对话历史 messages.append({ "role": "tool", "tool_call_id": tool_call.id, "content": function_result, }) # 5. 将函数执行结果送回给模型，让它生成面向用户的最终回答 second_response = client.chat.completions.create( model="claude-sonnet-4-6", messages=messages, ) final_message = second_response.choices[0].message messages.append(final_message) return final_message.content else: # 如果模型没有调用工具，直接返回其回复 return response_message.content # 6. 测试函数调用 if __name__ == "__main__": # 测试一个需要调用函数的查询 query1 = "北京现在的天气怎么样？用摄氏度告诉我。" print(f"用户: {query1}") answer1 = chat_with_function_calling(query1) print(f"助手: {answer1}\n") # 测试一个不需要调用函数的普通查询 query2 = "你好，请做一下自我介绍。" print(f"用户: {query2}") answer2 = chat_with_function_calling(query2) print(f"助手: {answer2}")

这个示例清晰地展示了函数调用的完整工作流：定义工具、模型请求调用、本地执行函数、返回结果、模型生成最终回答。通过这种方式，你可以将大模型与你的业务逻辑、内部数据或任何外部服务安全地连接起来。

4. 结合使用与注意事项

你可以将流式响应和函数调用结合起来。例如，在模型进行最终回答（即第二次调用，整合了函数结果后）时启用流式输出，这样用户就能看到模型“思考”后的流畅回复过程。

在使用这些进阶功能时，有几点需要注意：

• 流式响应与Token计数：在流式响应中，每个数据块（chunk）可能不包含完整的用量信息。总的使用量（如total_tokens）通常会在最后一个数据块中提供，你需要留意并妥善处理。
• 函数调用的成本：模型输出函数调用请求同样消耗Token。清晰、准确的函数描述有助于模型做出正确的调用决策，避免不必要的交互轮次。
• 错误处理：在实际生产环境中，务必为网络请求、流式中断、函数执行失败等情况添加健壮的错误处理与重试机制。
• 模型支持：并非所有模型都同等程度地支持函数调用。建议在Taotoken模型广场查看各模型的详细能力说明，或通过简单的测试进行验证。

通过本文的示例，你应该已经掌握了如何利用Taotoken平台，快速为你的Python应用添加流式响应和函数调用能力。这两种功能能显著提升应用的交互体验和智能化水平。更多详细的参数说明和高级用法，可以参考Taotoken平台的官方API文档。

开始构建你的智能应用吧，访问 Taotoken 创建API Key并探索更多模型。

🚀 告别海外账号与网络限制！稳定直连全球优质大模型，限时半价接入中。 👉 点击领取海量免费额度