截至目前,我已实际调用过市面上大部分主流的大模型 API。从使用体验来看,各类模型 API 调用大同小异,且有很大一部分直接兼容 OpenAI 的通用接口规范。不过由于大部分的产品文档写的不够简洁,个别模型的流式输出获取也存在差异,所以撰写本文,供大家参考。
尽管未来 API 的调用代码可能随版本更新而调整,但本文所涵盖的基础逻辑与实用技巧仍具有长期参考价值。并且,对于每个模型,文中均附带了官方文档链接,方便小伙伴随时查阅最新细节。
本文主打一个快速上手,力求让读者能快速入门并实践。
本文所提供的代码均通过实测,可以直接运行 。
获取 API_KEY
这个各个平台的路径都不同,但一般都是登录你的个人账户,然后在里面会有创建 API_KEY 的入口。记得要及时复制,因为关掉窗口后看不见了,只能重新创建一个了。
账号充值
部分平台有一定的免费额度,API 的调用原则上都是收费的,但是不贵,尤其是大厂的。
设置环境变量
在运行之前,要在系统环境变量中添加上你的 API_KEY,且设置好之后必须要重新启动一下 Jupyter,不然环境变量的修改不会生效。如果你实在不想设置的话,可以将下面代码中的“os.getenv(“”)”部分直接硬编码成你的 API_KEY,但是不太推荐!
Message 函数
def message_write(user_content, model=""): role_key = "Role" if model == "hunyuan" else "role" content_key = "Content" if model == "hunyuan" else "content" return [ {role_key: "system", content_key: " "}, {role_key: "user", content_key: user_content} ]
这个函数主要是用于后续复用,并处理腾讯混元模型的“大写”问题(就他和别人不一样,而且不大写还会报错!)
导入必要的库
import json import os import sys import types import time import requests from volcenginesdkarkruntime import Ark from openai import OpenAI from tencentcloud.common import credential from tencentcloud.hunyuan.v import hunyuan_client, models from zhipuai import ZhipuAI import sensenova import hashlib
这里主要是一些 API 的库,如果没有安装我先 pip install 一下。可以看到,除了字节(火山引擎)、腾讯、智谱和商汤的 sensenova,其他的基本都是用的 OpenAI 的接口。
接口文档:
ChatCompletions–火山方舟大模型服务平台-火山引擎
简明示例代码:
doubao_answer_model = "ep-015-ps7h6" def doubao_answer_x(question, knowledge_base, stream): client = Ark( api_key=os.environ.get("ARK_API_KEY"), #base_url="https://ark.cn-beijing.volces.com/api/v3" ) completion = client.chat.completions.create( model= doubao_answer_model, messages=message_write(question, "doubao"), stream=stream, ({'stream_options': {"include_usage": True}} if stream else {}), ) usage = None if stream: result = "" for chunk in completion: #print(chunk) if chunk.choices == []: usage = chunk.usage elif chunk.choices[0].delta.content: delta = chunk.choices[0].delta.content result = result + delta sys.stdout.write(delta) else: continue else: result = completion.choices[0].message.content usage = completion.usage sys.stdout.write(result) return result, usage doubao_answer_x( "你好,你叫什么名字?", False, True)
豆包与其他大模型最不同的一点是他的模型名称参数,需要登录火山引擎,创建自己的推理接入点,所以每个人的都是不一样的,以上仅是示例。
异步调用可以参考:ChatCompletions–火山方舟大模型服务平台-火山引擎
火山引擎的 API.tools 仅支持 function,不支持 retrieval,所以需要通过智能体调用知识库。可以参考下面这篇文章:
5千字长文:COZE专业版API调用超详细教程(Python)
扣子 ChatV3 文档:扣子-AI 智能体开发平台
但之前用下来,接入非豆包模型不太稳定,经常出现“ 远程主机强迫关闭了一个现有的连接”,希望现在有所改善了。
这里顺便说一句,非工业级产品的话,尽量不要用火山引擎的智能体,那个好贵,所以用 COZE 专业版就可以了。
接口文档:
通义千问API参考_大模型服务平台百炼(Model Studio)-阿里云帮助中心
简明示例代码:
qwen_answer_model = "qwen-plus" def qwen_answer_x(question, knowledge_base, stream): client = OpenAI( api_key=os.getenv("DASHSCOPE_API_KEY"), base_url="https://dashscope.aliyuncs.com/compatible-mode/v1", ) completion = client.chat.completions.create( model=qwen_answer_model, messages=message_write(question, "qwen"), stream=stream, ({'stream_options': {"include_usage": True}} if stream else {}), #temperature=0.8, #top_p=0.8 ) usage = None if stream: result = "" for chunk in completion: #print(chunk) if chunk.choices == []: usage = chunk.usage elif chunk.choices[0].delta.content: delta = chunk.choices[0].delta.content result = result + delta sys.stdout.write(delta) else: continue else: result = completion.choices[0].message.content usage = completion.usage sys.stdout.write(result) return result, usage qwen_answer_x("你好,你叫什么名字?", False, True)
和大部分大模型 API 一样,通义千问兼容 OpenAI 的接口规范。
异步调用可以参考:通义千问API参考_大模型服务平台百炼(Model Studio)-阿里云帮助中心
通义千问的 API.tools 也仅支持 function,不支持 retrieval,所以也需要通过智能体调用知识库。如果需要的话,我也可以准备一个分享教程。下面这个是官方文档:
通过API调用百炼应用_大模型服务平台百炼(Model Studio)-阿里云帮助中心
测试下来,相较于 COZE 的 RAG 返回,阿里云百炼的 RAG 返回似乎要准确得多。
接口文档:
腾讯混元大模型 对话_腾讯云
简明示例代码:
hunyuan_answer_model = "hunyuan-turbos-latest" def hunyuan_answer_x(question, knowledge_base, stream): secret_id = os.getenv("TencentCloud_SecretId") secret_key = os.getenv("TencentCloud_SecretKey") cred = credential.Credential(secret_id, secret_key) client = hunyuan_client.HunyuanClient(cred, "ap-beijing") 内地服务器 req = models.ChatCompletionsRequest() params = { "Model": hunyuan_answer_model, "Messages": message_write(question, "hunyuan"), "EnableEnhancement":False, 功能增强(如搜索)开关 "Stream": stream } req.from_json_string(json.dumps(params)) resp = client.ChatCompletions(req) if stream: result = "" if isinstance(resp, types.GeneratorType): # 流式响应 for event in resp: #print(event) delta = json.loads(event['data'])["Choices"][0]['Delta']['Content'] result = result + delta sys.stdout.write(delta 千问 Qwen 教程) usage = json.loads(event['data'])["Usage"] else: # 非流式响应 #print(resp) result = resp.Choices[0].Message.Content usage = resp.Usage sys.stdout.write(result) return result, usage hunyuan_answer_x("你好,你叫什么名字?", False, True)
腾讯混元也有 OpenAI 兼容接口,参考:腾讯混元大模型 混元 OpenAI 兼容接口相关调用示例_腾讯云
异步调用可以参考上面通义千问的 OpenAI 标准异步示例。
腾讯混元的 API.tools 也仅支持 function,不支持 retrieval,所以也需要通过智能体调用知识库,这里不进一步展开(因为体验下来豆包和通义的似乎在各个方面都略胜一筹)。
️ 以上三个是大厂的产品,相对而言更新频率比较稳定,回答质量较高,API 调用价格也相对低,首选推荐。下面即将提到的非大厂产品,各有各的特点,但最大的共性问题是资本实力不足,更新迭代无法保证,很多一开始出来是爆款,然后就渐渐开始走下坡路了。所以你要是做的是比较长期且比较重要的项目,建议还是不要选以下的产品。
接口文档:
首次调用 API | DeepSeek API Docs
简明示例代码:
deepseek_answer_model = "deepseek-chat" def deepseek_answer_x(question, knowledge_base, stream): client = OpenAI( api_key=os.getenv("DeepSeek_API_Key"), base_url="https://api.deepseek.com", ) completion = client.chat.completions.create( model=deepseek_answer_model, messages=message_write(question, "deepseek"), stream=stream, stream_options={"include_usage": True} #temperature=0.8, #top_p=0.8 ) if stream: result = "" for chunk in completion: if chunk.choices == []: usage = chunk.usage break elif chunk.choices[0].delta.content: #有个空值的结束标记,所以要判断 delta = chunk.choices[0].delta.content result = result + delta sys.stdout.write(delta) usage = chunk.usage else: result = completion.choices[0].message.content usage = completion.usage sys.stdout.write(result) return result, usage deepseek_answer_x("你好,你叫什么名字?", False, True)
非大厂的模型一般都会兼容 OpenAI 的接口,所以代码几乎一致。
值得一提的是,DeepSeek 的优势在深度推理,而其返回速度一直属于比较慢的,哪怕是 V3 Chat 模型。
接口文档:
基本信息 – Moonshot AI 开放平台 – Kimi 大模型 API 服务
简明示例代码:
kimi_answer_model = "moonshot-v1-auto" def kimi_answer_x(question, knowledge_base, stream): client = OpenAI( base_url="https://api.moonshot.cn/v1", api_key=os.environ["MOONSHOT_API_KEY"], ) completion = client.chat.completions.create( model=kimi_answer_model, messages=message_write(question, "kimi"), stream=stream, stream_options={"include_usage": True} #temperature=0.8, #top_p=0.8 ) if stream: result = "" for chunk in completion: if chunk.choices == []: usage = chunk.usage break elif chunk.choices[0].delta.content: #有个空值的结束标记,所以要判断 delta = chunk.choices[0].delta.content result = result + delta sys.stdout.write(delta) usage = chunk.usage else: result = completion.choices[0].message.content usage = completion.usage sys.stdout.write(result) return result, usage kimi_answer_x("你好,你叫什么名字?", False, True)
KIMI 可以通过 messages 传入文件作为知识库(全文阅读,不是召回片段),有兴趣的小伙伴可以参照这个:文件接口 – Moonshot AI 开放平台 – Kimi 大模型 API 服务
接口文档:
智谱AI开放平台
简明示例代码:
zhipu_answer_model = "GLM-4-plus" plus def zhipu_answer_x(question, knowledge_base_id=None, stream=True): if knowledge_base_id: print("调用知识库!") tools=[ { "type": "retrieval", "retrieval": { "knowledge_id": knowledge_base_id, "prompt_template": "从文档\n\"\"\"\n{{knowledge}}\n\"\"\"\n中找问题\n\"\"\"\n{{question}}\n\"\"\"\n的答案,找到答案就仅使用文档语句回答问题,找不到答案就用自身知识回答并且告诉用户该信息不是来自文档。\n不要复述问题,直接开始回答。" } } ] else: tools = [] client = ZhipuAI(api_key=os.getenv("ZhipuAI_API_KEY")) response = client.chat.completions.create( model=zhipu_answer_model, messages=message_write(question, "zhipu"), stream=stream, tools = tools, #max_tokens = 170, ) usage = None if stream: result = "" for chunk in response: delta = chunk.choices[0].delta.content result = result + delta sys.stdout.write(delta) usage = chunk.usage else: result = response.choices[0].message.content usage = response.usage sys.stdout.write(result) return result, usage zhipu_answer_x("你好,你叫什么名字?", 0, True)
智谱是少数几个用的不是 OpenAI 的兼容接口,他一开始的发展势头也确实比较猛。智谱的 API 可以叠加 retrieval 参数,让你调用上传到智谱平台的知识库(登录智谱 BigModel 控制台,上传知识库文件,就能获得知识库 id)。
接口文档:
百川大模型-汇聚世界知识 创作妙笔生花-百川智能
简明示例代码:
baichuan_answer_model = "Baichuan4-Turbo" def baichuan_answer_x(question, knowledge_base_ids=None, stream=True): if knowledge_base_ids: print("调用知识库!") extra_body={ "tools": [{ "type": "retrieval", "retrieval": { "kb_ids": knowledge_base_ids } }] } else: extra_body = {} client = OpenAI( api_key=os.environ.get("BaichuanAI_API_KEY"), base_url="https://api.baichuan-ai.com/v1/", ) completion = client.chat.completions.create( model=baichuan_answer_model, messages=message_write(question, "baichuan"), #temperature=0.3, stream=stream, extra_body=extra_body, ) if stream: result = "" for chunk in completion: result = result + chunk.choices[0].delta.content delta = chunk.choices[0].delta.content sys.stdout.write(delta) usage = chunk.usage else: result = completion.choices[0].message.content usage = completion.usage sys.stdout.write(result) if knowledge_base_ids: n_ref = 1 reference = "" if not stream: chunk = completion if chunk.knowledge_base.get('cites', []): for ref in chunk.knowledge_base.get('cites', []): if n_ref <=2: title = ref.get('title', []) text = ref.get('content', []) reference = reference + f"百川参考文档:[{n_ref}]\n{title}\n{text}\n\n" n_ref += 1 else: reference = "但是,百川没有返回参考文档。" print(reference) return result, usage baichuan_answer_x("你好,你叫什么名字?", ["kb-0yYosPLaSsVT7xCRYA9rPBw5"], True)
百川同样可以召回在线知识库(同样是登录账号后,上传知识库文件,获得知识库 id)。并且他可以返回具体的召回内容,这里就增加了一段代码来显示召回内容。不过,使用下来,百川知识库调用的回答效果并不理想。
接口文档(需要注册登录):
SenseCore-商汤大装置-帮助中心
简明示例代码:
sensenova_answer_model = "SenseChat-5" def sensenova_answer_x(question, knowledge_base_ids, stream): if knowledge_base_ids: print("调用知识库!") know_ids = knowledge_base_ids knowledge_config = { "control_level": "normal", #设置为high时模型回答尽可能依赖知识库,不自由发散。 "knowledge_base_result": True, #是否知识库检索结果 "knowledge_base_configs":[ { "know_id": knowledge_base_ids[0], "faq_threshold": 0.3, #知识库中的qa_lst精准命中程度的阈值配置,越高相似度要求越严格 }, ], } else: know_ids = [] knowledge_config = {} resp = sensenova.ChatCompletion.create( messages=message_write(question, "sensecore"), model=sensenova_answer_model, know_ids = know_ids, stream=stream, knowledge_config=knowledge_config, #max_new_tokens=1024, #temperature=0.8, #top_p=0.7, #user="sensenova-python-test-user", ) if not stream: resp = [resp] result = "" for part in resp: choices = part['data']["choices"] usage = part['data']["usage"] #print(choices) for c_idx, c in enumerate(choices): if len(choices) > 1: sys.stdout.write("===== Chat Completion {} =====\n".format(c_idx)) if stream: delta = c.get("delta") if delta: result = result + delta sys.stdout.write(delta) else: result = c["message"] sys.stdout.write(result) if len(choices) > 1: sys.stdout.write("\n") sys.stdout.flush() if knowledge_base_ids: n_ref = 1 reference = "" if part['data']["knowledge_base_results"]: for ref in part['data']["knowledge_base_results"][0]["results"]: if n_ref <=2: text = ref["result"] reference = reference + f"商汤参考文档:[{n_ref}]\n{text}\n\n" n_ref += 1 else: reference = "但是,商汤没有召回任何参考文档。" print(reference) return result, usage sensenova_answer_x("你好,你叫什么名字?", ["se2969afc13b4449dbca9342edc14e8f5"], True)
不得不说,商汤的反馈速度是这里所有 API 中最快的。知识库召回的结果也还行,是目前体验下来还不错的 API。不过他代码逻辑也是和其他几个相差最大的一个,需要注意。
接口文档:
文档中心 – MiniMax 开放平台
简明示例代码:
minimax_answer_model = "abab6.5s-chat" def minimax_answer(question, knowledge_base, stream): client = OpenAI( api_key=os.getenv("MiniMax_API_KEY"), base_url="https://api.minimax.chat/v1", ) completion = client.chat.completions.create( model=minimax_answer_model, messages=message_write(question, "minimax"), stream=stream, #max_tokens=, #temperature=0.3, #top_p=0.9 ) if stream: result = "" for chunk in completion: if chunk.choices[0].delta.content: #有个空值的结束标记,所以要判断 delta = chunk.choices[0].delta.content result = result + delta sys.stdout.write(delta) usage = chunk.usage else: result = completion.choices[0].message.content usage = completion.usage sys.stdout.write(result) return result, usage minimax_answer("你好,你叫什么名字?", False, True)
完全标准的接口代码。MiniMax 的效果总体比较拉胯,也是这里唯一一个不知道自己名字、也不能返回 usage 的大语言模型,不值得充钱。
接口文档:
天工开放平台
简明示例代码:
def sky_answer_x(question, knowledge_base, stream): url = 'https://api-maas.singularity-ai.com/sky-work/api/v1/chat' app_key = os.getenv("sky_app_key") app_secret = os.getenv("sky_app_secret") timestamp = str(int(time.time())) sign_content = app_key + app_secret + timestamp sign_result = hashlib.md5(sign_content.encode('utf-8')).hexdigest() headers={ "app_key": app_key, "timestamp": timestamp, "sign": sign_result, "Content-Type": "application/json", } data = { "messages": message_write(question, "sky"), "intent": "chat", # 默认为空将进行意图识别判定是否搜索增强,取值 'chat'则不走搜索增强 } response = requests.post(url, headers=headers, json=data, stream=True) 只能流式 result = "" for line in response.iter_lines(): line = line.decode('utf-8')[6:] #print(line) if line and line!="[DONE]": parsed_data = json.loads(line) if parsed_data["card_type"]=="markdown": delta = parsed_data['arguments'][0]['messages'][0]['text'] result = result +delta if stream: sys.stdout.write(delta) if not stream: print(result) return result sky_answer_x("你好,你叫什么名字?", False, True)
不兼容 OpenAI 接口,天工 sky 的速度也是最慢的(和 DeepSeek 有的一拼),也不太推荐。
其他的诸如阶跃星辰、零一万物的接口也大同小异,就不一一列举了,希望他们能撑得下去吧。
发布者:Ai探索者,转载请注明出处:https://javaforall.net/259768.html原文链接:https://javaforall.net
