计算岗位匹配度

@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)
    }

    @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

先把投递记录捞出来，靠它拿到 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}")

核心流程：发问题 → 模型说要调工具 → 执行工具 → 把结果塞回 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_mapping[func_name](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_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}")

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

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

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

三块代码的整体关系

• 第一块是业务接口：把简历和岗位喂给大模型，让它算匹配度，本质是"单轮 prompt，不要工具"。

• 第二、三块是 Agent 工具调用的演示：模型能自己决定调哪个工具、传什么参数，拿到结果再总结。区别是单模型版一次只处理一个 tool_call，多模型版能并发处理多个。

• 共同点：都走百炼兼容模式地址、都用 qwen3.7-plus；工具调用时 tool_call_id 一定不能丢。