计算岗位匹配度
@job_router.get(“/resume_submission_detail/{id}”, summary=“简历投递详情”)
async def resume_submission_detail(id:int):
res=await JobService.resume_submission_detail(id)
return {
“code”: 1,
“message”: “查询成功”,
“data”: json.loads(res)
}
Lobster AI
python
运行
@staticmethod
async def resume_submission_detail(id: int):
resume_submission_records =await ResumeSubmissionRecords.get_or_none(id=id)

#职位信息 job = await Job.get_or_none(id=resume_submission_records.job_id) #求职者信息 job_seeker = await JobSeeker.get_or_none(id=resume_submission_records.job_seeker_id) #简历基本信息 resume_basic_info = await ResumeBasicInfo.get_or_none(job_seeker_id=job_seeker.id) job_seeker_age = now().year - resume_basic_info.birth_date.year #求职意向 job_intention = await JobIntention.get_or_none(job_seeker_id=job_seeker.id) #工作经历 work_experiences = await WorkExperience.filter(job_seeker_id=job_seeker.id) work_experiences_list=[] for i in work_experiences: work_experience_data={ "公司名称":i.company_name, "所属行业":i.industry, "职位名称":i.company_name, "入职时间":i.entry_time, "离职时间":i.leave_time, "工作内容":i.work_content, "工作业绩": i.performance, "拥有技能列表":i.skills, } work_experiences_list.append(work_experience_data) #项目经历 project_experiences=await ProjectExperience.filter(job_seeker_id=job_seeker.id) project_experiences_list=[] for j in project_experiences: project_experience_data={ "项目名称": j.project_name, "项目角色": j.project_role, "项目开始时间": j.start_time, "项目结束时间": j.end_time, "项目描述": j.project_desc, "项目业绩": j.project_performance, } project_experiences_list.append(project_experience_data) #教育经历 education_experiences=await EducationExperience.filter(job_seeker_id=job_seeker.id) education_experiences_list=[] for k in education_experiences: education_experience_data={ "学历名称": k.education_name, "学制类型": k.study_type, "学校名称": k.school_name, "专业名称": k.major_name, "入学时间": k.entry_date, "毕业时间": k.graduate_date, "在校经历": k.school_experience, } education_experiences_list.append(education_experience_data) #优势 advantage_evaluation = await AdvantageEvaluation.filter(job_seeker_id=job_seeker.id) advantage_list = [] for advantage in advantage_evaluation: advantage_data={ "个人优势": advantage.personal_advantage, "自我评价": advantage.self_evaluation, } advantage_list.append(advantage_data) #专业技能 professional_skill = await ProfessionalSkill.filter(job_seeker_id=job_seeker.id) professional_skill_list = [] for skill in professional_skill: professional_skill_data={ "技能描述": skill.skill_desc } professional_skill_list.append(professional_skill_data) prompt = f""" ##角色设定： 你是一个经验丰富的人力资源专家 ##任务描述： 根据求职者的的简历内容和岗位的职位描述，分析计算岗位匹配度和理由 ##输入数据： 1.岗位的职位描述： 1.1.职位名称：{job.job_name} 1.2.工作地点：{job.work_location} 1.3.最低薪资：{job.min_salary} 1.4.最高薪资：{job.max_salary} 1.5.经验要求：{job.exp_require} 1.6.学历要求：{job.edu_require} 1.7.性别要求：{job.gender_require} 1.8.职位描述：{job.job_desc} 1.9.任职要求：{job.duty_require} 2.求职者的的简历内容 2.1.求职者性别（1-男，2-女）：{resume_basic_info.gender} 2.2.求职者年龄：{job_seeker_age} 2.3.求职者婚姻状况（1-未婚 2-已婚 3-保密）：{resume_basic_info.marital_status} 2.4.求职者政治面貌：{resume_basic_info.political_status} 2.5.求职者期望职位：{','.join([i.get("name") for i in job_intention.expect_position])} 2.6.求职者期望薪资：{job_intention.expect_salary} 2.7.求职者工作性质（1-全职 2-兼职 3-其他）：{job_intention.work_type} 2.8.求职者期望城市：{",".join([i.get("name") for i in job_intention.expect_city])} 2.9.求职者期望行业：{",".join([i.get('name') for i in job_intention.expect_industry])} 2.10.求职者工作经历：{work_experiences_list} 2.11.求职者项目经历：{project_experiences_list} 2.12.求职者教育经历：{education_experiences_list} 2.13.求职者技能：{professional_skill_list} 2.14.求职者优势：{advantage_list} ##输出格式： 以JSON格式输出，包含以下字段： job_matching_degree：岗位匹配度（0-100）百分比，分数越高，匹配度越高 matching_reason：岗位匹配理由，尽可能每个方面都分析到 ##约束条件： 1.请根据输入数据，分析计算岗位匹配度和理由，并输出结果 2.请使用中文描述 3.只输出JSON格式，且字段必须按照指定格式输出，不要输入其他内容 ##输出示例： {{“job_matching_degree”:“78%”,"matching_reason": "求职者期望职位列表与岗位职位描述匹配度高:项目经历匹配"}} """ client = OpenAI( # 若没有配置环境变量，请用百炼API Key将下行替换为：api_key="sk-xxx" api_key=os.getenv("DASHSCOPE_API_KEY"), base_url="https://ws-5p10gbdo8vadjgq7.cn-beijing.maas.aliyuncs.com/compatible-mode/v1", ) completions = client.chat.completions.create( model="qwen3.7-plus", messages=[{"role":"user","content":prompt}] ) res = completions.choices[0].message.content return res

Lobster AI
python
运行

先把投递记录捞出来，靠它拿到 job_id 和 job_seeker_id —— 这俩是后面所有查询的入口

年龄这块是用出生年份减当前年份算的，够糙但够用（没管具体月份，差个把月无所谓）

下面是求职意向和三段经历（工作 / 项目 / 教育），每个都查出来拼成 list，方便后面整段塞进 prompt 给模型 看。

项目经历、教育经历同理，都是查出来转成字典列表

优势和技能这两段，原代码逻辑一样，也是逐条转字典

拼 prompt 调模型。这里期望职位 / 城市 / 行业是列表（每个元素是带 name 的字典），所以用 “,”.join(…) 拼成字符串，让模型读着舒服，这里是提示词工程，要着重注意提示词 工程的六要素。

调用外部工具

大模型本身并没有这些实时数据，想要获取实时数据就只能通过两种方式，爬虫(获取公开的数据)和调用外部工具（Function Calling），在此处的例子是调用了外部工具去获取真实的实时数据，由上图可以看出，大模型是被调用了两次，一次用来判断是否需要调用工具，一次是给出回复（在这之前其实已经是有结果了，此处只是输出更友好的，优化后的结果）

单模型调用测试
模拟一个查天气的小工具，看模型怎么决定"要不要调工具、调完怎么总结"。

先定义工具（告诉模型有这么个工具、要什么参数）：

#定义工具
tools=[
{
“type”: “function”,
“function”: {
“name”: “get_current_weather”,
“description”: “当你想查询指定城市的天气时非常有用。”,
“parameters”: {
“type”: “object”,
“properties”: {
“location”: {
“type”: “string”,
“description”: “城市或县区，比如北京市、杭州市、余杭区等。”,
}
},
“required”: [“location”],
},
},
},
]

#模拟天气查询工具
def get_current_weather(arguments):
#arguments 形参 字典
weather_conditions = [“晴天”, “多云”, “雨天”]
random_weather = random.choice(weather_conditions)
location = arguments[“location”]
return f"{location}今天是{random_weather}。"

#获取AI的回复
def get_ai_response(messages):
completion = client.chat.completions.create(
model=“qwen3.7-plus”,
messages=messages,
tools=tools,
)
return completion

USER_QUESTION = “北京天气咋样”

messages = [{“role”: “user”, “content”: USER_QUESTION}]
response = get_ai_response(messages)
assistant_output = response.choices[0].message
if assistant_output.content is None:
assistant_output.content = “”
messages.append(assistant_output)

如果不需要调用工具，直接输出内容

if assistant_output.tool_calls is None:
print(f"无需调用天气查询工具，直接回复：{assistant_output.content}“)
else:
# 进入工具调用循环
while assistant_output.tool_calls is not None:
tool_call = assistant_output.tool_calls[0]
tool_call_id = tool_call.id
func_name = tool_call.function.name
arguments = json.loads(tool_call.function.arguments)
print(f"正在调用工具 [{func_name}]，参数：{arguments}”)
# 执行工具
tool_result = get_current_weather(arguments)
# 构造工具返回信息
tool_message = {
“role”: “tool”,
“tool_call_id”: tool_call_id,
“content”: tool_result, # 保持原始工具输出
}
print(f"工具返回：{tool_message[‘content’]}“)
messages.append(tool_message)
# 再次调用模型，获取总结后的自然语言回复
response = get_ai_response(messages)
assistant_output = response.choices[0].message
if assistant_output.content is None:
assistant_output.content = “”
messages.append(assistant_output)
print(f"助手最终回复：{assistant_output.content}”)
Lobster AI
python
运行

核心流程：发问题 → 模型说要调工具 → 执行工具 → 把结果塞回 messages → 再让模型用大白话总结。

在这里需要格外注意代码： tool_call = assistant_output.tool_calls[0]

因为此处是单个工具调用，可以用下标为0的索引去取，如果是可以调用多个工具的话，在此处是不可以这样用的，此处是个列表，若是调用多个模型的话就只能用遍历的方法来取。如下面的例子

多模型调用测试
和第二套思路一样，但定义了两个工具（天气 + 查学历），而且支持模型一次返回多个 tool_calls 一起跑

模拟用户问题

USER_QUESTION = “帮我查询一下学历, 验证码是:A…”

定义工具列表

tools = [
{
“type”: “function”,
“function”: {
“name”: “get_current_weather”,
“description”: “当你想查询指定城市的天气时非常有用。”,
“parameters”: {
“type”: “object”,
“properties”: {
“location”: {
“type”: “string”,
“description”: “城市或县区，比如北京市、杭州市、余杭区等。”,
}
},
“required”: [“location”],
},
},
},
{
“type”: “function”,
“function”: {
“name”: “call_api_get”,
“description”: “当你想查询查询学历或者验证学历时非常有用。”,
“parameters”: {
“type”: “object”,
“properties”: {
“vcode”: {
“type”: “string”,
“description”: “学历验证码。”,
}
},
“required”: [“vcode”],
},
},
},
]

模拟天气查询工具

def get_current_weather(arguments):
weather_conditions = [“晴天”, “多云”, “雨天”]
random_weather = random.choice(weather_conditions)
location = arguments[“location”]
return f"{location}今天是{random_weather}。"

API_KEY = “MY_XXW_API_KEY”

def call_api_get(arguments):
vcode=arguments[“vcode”]
BASE_URL = “https://www.apimy.cn/api/xxw/bgcx”
key=f"boss:llm:academic_credential_verification:{vcode}"
redis_data=redis_client.get(key)
if redis_data is None:
payload = {“key”: os.getenv(“自己的环境变量”), “vcode”:vcode}
headers = {“Content-Type”: “application/json”}
response = requests.post(BASE_URL, json=payload, headers=headers, timeout=30)
response.raise_for_status()
data = response.json()
redis_client.set(key, json.dumps(data, ensure_ascii=False))

return json.dumps(data, ensure_ascii=False) else: return redis_data

封装模型响应函数

def get_response(messages):
completion = client.chat.completions.create(
model=“qwen3.7-plus”,
extra_body={“enable_thinking”: False},
messages=messages,
tools=tools,
)
return completion

messages = [{“role”: “user”, “content”: USER_QUESTION}]
response = get_response(messages)
assistant_output = response.choices[0].message
if assistant_output.content is None:
assistant_output.content = “”
messages.append(assistant_output)

如果不需要调用工具，直接输出内容

if assistant_output.tool_calls is None:
print(f"无需调用查询工具，直接回复：{assistant_output.content}“)
else:
# 进入工具调用循环
while assistant_output.tool_calls is not None:
tool_call = assistant_output.tool_calls
for i in tool_call:
tool_call_id = i.id
func_name = i.function.name
arguments = json.loads(i.function.arguments)
print(f"正在调用工具 [{func_name}]，参数：{arguments}”)
function_mapping ={
“get_current_weather”: get_current_weather,
“call_api_get”: call_api_get
}
# 执行工具
tool_result =function_mappingfunc_name
# 构造工具返回信息
tool_message = {
“role”: “tool”,
“tool_call_id”: tool_call_id,
“content”: tool_result, # 保持原始工具输出
}
print(f"工具返回：{tool_message[‘content’]}“)
messages.append(tool_message)
# 再次调用模型，获取总结后的自然语言回复
response = get_response(messages)
assistant_output = response.choices[0].message
if assistant_output.content is None:
assistant_output.content = “”
messages.append(assistant_output)
print(f"助手最终回复：{assistant_output.content}”)
Lobster AI
python
运行

学历查询这边带了个 redis 缓存，验证码当 key，避免对同一个验证码重复打外部 API

模型调用时关掉了思考模式（enable_thinking=False）——工具调用场景要快、要干脆，不需要模型在那长篇推理

多工具循环：用 for i in tool_call 把模型返回的每一个 tool_call 都跑一遍，再统一回喂模型总结

三块代码的整体关系
第一块是业务接口：把简历和岗位喂给大模型，让它算匹配度，本质是"单轮 prompt，不要工具"。
第二、三块是 Agent 工具调用的演示：模型能自己决定调哪个工具、传什么参数，拿到结果再总结。区别是单模型版一次只处理一个 tool_call，多模型版能并发处理多个。
共同点：都走百炼兼容模式地址、都用 qwen3.7-plus；工具调用时 tool_call_id 一定不能丢。