doubao_api.py
doubao_api.py — 豆包大模型接口（核心）
call_gpt / call_gpt_system 双接口；fix_return_json 清理大模型返回内容中的 Markdown json 代码围栏包裹符，保证返回纯 JSON。业务逻辑最重：提取职位 20-30 个维度的精细字段。
代码
# -*- coding: utf-8 -*-
# @Time : 2025/2/17 16:39
# @Author : chang
# 导入时间模块，用于延时、时间相关操作
import time
# 导入火山引擎方舟大模型 SDK
from volcenginesdkarkruntime import Ark
# 导入系统模块，用于修改 Python 路径
import sys
# 将上级目录加入模块搜索路径，方便导入工具类
sys.path.append('../')
# 导入日志工具，用于记录运行日志、错误日志
from utils import ner_logger
# 导入邮件发送工具，用于异常时发送告警邮件
from utils_log import send_email
# 火山引擎方舟 API Key（身份凭证）
api_key = "99458ecf-4a8e-4e2a-ad96-246caf5fd213"
# 调用的大模型 ID（当前使用的模型）
model_id = "ep-20250613142735-f4xrw"
# 旧模型 ID（已注释，备用）
# model_id = "ep-20250219151347-fspmw"
# 模型名称拼接，用于日志标识
model_name = f"model:deepseek ,model_id:{model_id}"
def fix_return_json(_text):
"""
清理大模型返回内容中的 Markdown 格式符号
主要用于去掉 ```json ``` 包裹符，保证返回纯 JSON 字符串
"""
# 去掉开头的 ```json
_text = _text.replace("```json", "")
# 去掉结尾的 ```
_text = _text.replace("```", "")
return _text
def call_gpt(prompt, isjson=False) -> str:
"""
调用 deepseek 大模型接口（基础版）
:param prompt: 输入给 AI 的提示词
:param isjson: 是否要求返回 JSON 格式
:return: (是否成功, 返回内容)
"""
# 提交请求
try:
# 记录原始请求日志
ner_logger.info(f"doubao-deepseek old:\n {prompt}")
# 初始化火山引擎方舟客户端
client = Ark(api_key=api_key)
# 构造消息体：系统角色 + 用户提问
messages = [{"role": "system", "content": "你是根据用于分析招聘数据、提取结构化信息的AI机器人"}]
messages.append({"role": "user", "content": prompt})
# 调用大模型聊天接口
completion = client.chat.completions.create(
model=model_id, # 模型ID
messages=messages, # 对话消息
temperature=0, # 温度（0=最稳定、不发散）
max_tokens=12288, # 最大输出长度
top_p=0.2, # 核采样，控制随机性
frequency_penalty=0, # 频率惩罚
presence_penalty=0, # 存在惩罚
stop=None # 停止符
)
# 将返回对象转为字典，方便取值
completion = completion.to_dict()
# 提取 AI 返回的内容
content = completion["choices"][0].get("message").get("content")
# 清理返回内容中的 Markdown 符号
_value_str = fix_return_json(content)
# 返回成功 + 结果
return True, _value_str
except Exception as e:
# 捕获异常，打印堆栈
import traceback
traceback.print_exc()
# 记录错误日志
ner_logger.error(f"doubao error:{e}")
# 发送错误告警邮件
send_email(f"doubao error:{e}<br>{prompt}")
# 最终失败返回
return False, "大模型处理失败[openai][48]"
def call_gpt_system(prompt, isjson=False) -> str:
"""
调用 deepseek 大模型接口（系统提示词增强版）
专门用于招聘数据抽取，强制 JSON 输出
:param prompt: 提示词
:param isjson: 是否 JSON 格式
:return: (是否成功, 返回内容)
"""
# 组建消息体：系统角色（更严格）
message_text = [{"role": "system", "content": "你是用于分析招聘数据、提取结构化信息的AI机器人，帮助用户分析请求的数据，返回符合要求的json格式的数据"}]
# 添加用户输入
message_text.append({"role": "user", "content": prompt})
# 记录请求日志
ner_logger.info(f"{model_name}:\n {prompt}")
# 初始化返回值
_value_str = ""
# 提交请求
try:
# 初始化客户端
client = Ark(api_key=api_key)
# 调用模型
completion = client.chat.completions.create(
model=model_name,
messages=message_text,
response_format={"type": "json_object"} if isjson else None, # JSON 格式控制
temperature=0,
max_tokens=12288,
top_p=0.1,
frequency_penalty=0,
presence_penalty=0,
stop=None,
timeout=60 * 6 # 超时时间 6 分钟
)
# 转字典
completion = completion.to_dict()
# 获取返回结果
choices = completion.get("choices")
if choices:
message = choices[0].get("message")
if message:
_value_str = message.get("content")
# 记录返回结果
ner_logger.info(f"{model_name} return :\n{_value_str}\n")
# 清理格式
_value_str = fix_return_json(_value_str)
return True, _value_str
except Exception as e:
# 异常捕获
import traceback
traceback.print_exc()
ner_logger.error(f"openai error:{e}")
# 失败返回
return False, "大模型处理失败[openai][79]"
def getVers():
"""
获取当前使用的模型版本信息
"""
return model_name
# ====================== 测试用提示词（职位信息抽取） ======================
demo_str = '''
# 指令
我将提供一段公司官网“职位信息”的内容，你需要从中提取以下信息，并输出为JSON格式:
1.职位名/岗位名称(JobTitle)
2.发布日期/开始日期(PublishTime)
3.到期日期/截止日期/结束日期(CutDate)
4.学历(Degree)：可能是多个，不重复，是个数组，可选择：“大专”，“大专及以上”、“本科”、“本科及以上”、“硕士”、“硕士/MBA”、“研究生”、“硕士及以上”、“研究生及以上”、“博士”
5.年龄要求(Age)：对候选人的岁数要求如：“18-25岁”、“40岁及以上”等
6.职位薪资(Salary)：职位的薪资情况，如：“面议”、“6k-8k”、“8k-10k”等
7.招聘人数(JobNum)：职位招聘的具体人数，如：“1人”等
8.工作年限(WorkYears)：工作经历具体年限要求，如：“1年及以上”、“5-8年”、“10年以下”等，如果是中文数字如一二三等转换成阿拉伯数字，只要年限取大的即可，无经验为空
9.专业要求(MajorRequirement)
10.工作地点(WorkPlace)：岗位所在城市、工作地区、公司所在地、工作地点、职场坐标、国外国家及地区，可能是多个，用逗号分隔，不要输出数组
11.详细地址(Address)
12.工作类型(HopeWorkType)：可选择：全职/兼职/实习等，注意职位要求的经验不能当工作类型
13.所属部门/工作部门/公司(JobDept)
14.职位描述/职位职责/工作职责/岗位职责(JobDescribe),保留原文中的换行符号,(职位描述/职位职责/工作职责/岗位职责)都为JobDescribe
15.职位要求/工作要求/任职要求/任职资格(Jobreq),保留原文中的换行符号，(职位要求/工作要求/任职要求/任职资格)都为Jobreq
16.暑期实习(SummerInternship)：可选择：有/无
17.转正机会(RegularEmployee) ： 可选择：有/无
18.职业类别(JobCategory)： 职位为的分类，如 职能 、技术、运营、市场、人事、财务、行政、客服、销售、市场、人事、财务、行政、客服、销售、市场等
19.技能标签(Skills)：提取最多5个关键词，如：python、java、大数据等
20.行业要求(IndustryRequirement)
21.语言要求(LanguageRequirement)
21.证书要求(CertificateRequirement)
22.工作时间(WorkTime)
23.工资结算方式(SalaryPayment)
24.院校要求(SchoolRequirement)
#Rules：
1. 职位信息中没有提及的信息，输出空即可；
2. 完全按照原文输出，不需要加工。
3. 你所需要的全部内容都在[webpage X begin]...[webpage X end]中，不能虚构内容。
4. 如果存在多职位，那么只返回第一个职位信息即可
5. 今天是2025-04-29
[webpage X begin]
SSC实习生8013广州市-天河区2025-04-22发布大专无经验招聘1人工作职责1、员工档案归档：负责人事档案扫描和录入工作，按要求完成档案扫描及装订归档，并录入系统；协助归档离职员工资料，并按要求打包存库；
2、入职手续办理：协助办理员工入职手续的系统录入；
3、系统信息整理：协助日常人事档案材料收集、建立，完善HRIS系统人事信息；
4、团队协作：其他临时性事项处理、支持。任职要求1、学历：大专以上学历，档案管理、中文、人力资源、文秘或相关专业
2、相关工作经验优先
[webpage X end]
'''
if __name__ == '__main__':
# 测试调用大模型
ask = call_gpt(demo_str)
print(ask)
项目分区导航：deepseek_api ⬅️ | 02-doubao_api | ➡️ doubao_api_new
💬 评论