调用外部工具接口

调用学历验证接口

这块的核心思路是：用户说话 → 先让一个小模型把"验证码"抠出来 → 有验证码就走工具调用去查学历 → 没验证码就直接回一句提示。

client = OpenAI(
    # 若没有配置环境变量，请用百炼API Key将下行替换为：api_key="sk-xxx"
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="httle-mode/v1",
)

# 封装模型响应函数
def get_ai_response(messages):
    completion = client.chat.completions.create(
        model="qwen3.7-plus",
        extra_body={"enable_thinking": False},
        messages=messages,
        tools=tools,
    )
    return completion


def call_api_get(arguments):
    if arguments is None:
        return "请提供正确的学历验证码"
    vcode=arguments["vcode"]
    if vcode is None:
        return "请提供正确的学历验证码"
    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("MY_XXW_API_KEY"), "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


tools = [
    {
        "type": "function",
        "function": {
            "name": "call_api_get",
            "description": "当你想查询查询学历或者验证学历时非常有用。",
            "parameters": {
                "type": "object",
                "properties": {
                    "vcode": {
                        "type": "string",
                        "description": "学历验证码，通常由大写字母和数字组成，例如A54TFQ。请从用户消息中提取该验证码。",
                    }
                },
                "required": ["vcode"],
            },
        },
    },
]


# 1. 独立结构化提取Prompt：专门从用户query提取验证码
EXTRACT_VCODE_PROMPT = """
你是文本提取器，只输出严格JSON，无多余文字。
任务：从用户输入语句中提取学历验证码vcode。
规则：
1. 如果用户没有提供任何验证码，vcode字段填空字符串""
2. 如果存在字母、数字组合的学历验证码，完整提取到vcode
输出格式固定：{"vcode": "提取到的验证码或空字符串"}
"""

# 2. 业务主系统prompt
MAIN_SYSTEM_PROMPT = """
你是学历核验助手：
1. 用户提供验证码，调用academic_credential_verification工具核验学历；
2. 用户未提供验证码，直接回复：请提供正确的学历验证码；
"""


# 封装通用LLM调用
def llm_chat(messages, temperature=0.7, tools=None):
    params = {
        "model": "qwen3.7-plus",
        "messages": messages,
        "temperature": temperature
    }
    if tools:
        params["tools"] = tools
    return client.chat.completions.create(**params)

# 结构化提取验证码函数
def extract_vcode_from_user_input(user_query: str) -> str:
    extract_msg = [
        {"role": "system", "content": EXTRACT_VCODE_PROMPT},
        {"role": "user", "content": user_query}
    ]
    resp=llm_chat(extract_msg,temperature=0.1)
    content=resp.choices[0].message.content.strip()
    try:
        res_json=json.loads(content)
        vcode=res_json.get("vcode","")
        logger.info(f"结构化提取结果，输入：{user_query}，提取vcode：{vcode}")
        return vcode
    except Exception as e:
        logger.error(f"提取验证码解析失败: {str(e)}, raw_content: {content}")
        return ""


@llm_day04_router.post("/function_calling_api", summary="函数调用API")
async def function_calling_api(request: LLMCase3):
    user_query = request.user_query
    # 第一步：结构化LLM提取验证码
    vcode = extract_vcode_from_user_input(user_query)
    # 无验证码，直接返回提示，不走工具调用流程
    if not vcode:
        return {
            "code": 1,
            "message": "调用成功",
            "data": {
                "result": "请提供正确的学历验证码"
            }
        }
    # 有验证码，初始化全新消息上下文（修复全局messages污染问题）
    messages = [{"role": "system", "content": MAIN_SYSTEM_PROMPT}]
    messages.append({"role": "user", "content": user_query})

    #第一轮
    completion = llm_chat(messages, tools=tools)
    ai_msg = completion.choices[0].message
    messages.append(ai_msg)

    if not ai_msg.tool_calls:
        logger.info("本次调用不需要走function calling")
        return {
            "code": 1,
            "message": "调用成功",
            "data": {
                "result": ai_msg.content
            }
        }

    logger.info("本次调用需要走function calling")
    func_mapping = {
        "academic_credential_verification": call_api_get
    }
    # 执行工具调用
    for tool_call in ai_msg.tool_calls:
        tool_id = tool_call.id
        func_name = tool_call.function.name
        func_args = json.loads(tool_call.function.arguments)
        tool_result = func_mapping[func_name](func_args)
        messages.append({
            "role": "tool",
            "tool_call_id": tool_id,
            "content": tool_result
        })
    # 工具结果回传给模型生成最终回答
    final_completion = llm_chat(messages, tools=tools)
    final_ans = final_completion.choices[0].message.content
    logger.info(f"本次问答最终的结果:{final_ans}")
    return {
        "code": 1,
        "message": "调用成功",
        "data": {
            "result": final_ans
        }
    }

call_api_get（）：这是被模型调用的真实函数。先防御空参数，避免后面崩溃；再用验证码当 redis key 查缓存；没命中才打外部接口，查完写缓存下次直接命中；两种情况都返回 JSON 字符串作为工具结果。缓存 + 入参防御是这段的精髓：省钱又稳。

两个prompt：

把任务拆成两个独立 prompt。第一个是纯提取器，强制只输出 {"vcode": "..."}，职责单一好解析；第二个是业务助手人设，管流程。两者分离，互不污染，温度也能分开调。

 extract_vcode_from_user_input 用低温度（0.1）调提取器，提高稳定性；解析用 try/except 包住，失败也不崩，返回空串交给主流程处理——这种"降级而非崩溃"的写法很实用。

接口主体：

链路编排：扣验证码 → 空则直接提示 → 有则起干净上下文 → 首轮让模型决定调不调工具 → 不调就直接返回、调就遍历 tool_calls 执行并把结果按 tool_call_id 塞回对话 → 再调一次模型生成最终答复。 tool_call_id 必须带上，模型靠它对账。每步有日志，好排查。

RAG语义检索初识

整体：读文档 → 切片 → 转向量 → 存 Chroma；提问时把问题也转向量召回最相关片段，拼进 prompt 让模型基于片段作答。

def load_file(file_path):
    document=Document(file_path)

    all_text=[]

    for i in document.paragraphs:
        all_text.append(i.text)

    return "\n".join(all_text)



#分片
def split_text(all_text):
    chunks=[]
    for j in range(0,len(all_text),200):
        chunks.append(all_text[j:j+200])
    return chunks

client = OpenAI(
    # 若没有配置环境变量，请用百炼API Key将下行替换为：api_key="sk-xxx"
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="ht1",
)


#文本转换向量
def text_embedding(chunks):
    #对于文档的内容和进行了分片，因此是一个列表，但是对于用户来说，用户输入的是单个问题，是一个字符串，因此有两种处理方式，
    #第一种是在此处将用户的问题转换为列表无非就是加一个判断，第二种是在将用户问题进行检索的时候转换为列表。
    #第二种虽然更显方便，但是更土建第一种，这样就不需要在后面判断输入的类型了。这样更贴合实际，更像是将text_embedding封装成了一个工具类
    if isinstance(chunks,str):
        chunks = [chunks]
    else:
        chunks=chunks

    step=10
    embeddings_list=[]
    for i in range(0,len(chunks),step):
        batch=chunks[i:i+step]
        embeddings=client.embeddings.create(
            model="text-embedding-v4",
            input=batch
        )
        batch_embeddings = [j.embedding for j in embeddings.data]
        embeddings_list.extend(batch_embeddings)
    return embeddings_list


#初始化chroma的客户端
CLIENT = chromadb.PersistentClient()

#将文本向量存入到向量数据库中（chroma）
def save_to_vector_db(chunks):
    collection = CLIENT.get_or_create_collection(
        name="employee_manual_collection",
        embedding_function=None)                        #不用自带的文件切分
    collection.add(
        ids=[f"chunl_{k}" for k in range(len(chunks))],  #ids集合
        documents=chunks,                                #文本集合
        embeddings=text_embedding(chunks),               #向量集合
    )


#根据用户输入的query进行检索
def query_by_chroma(user_query_text):
    user_query_embeddings=text_embedding(user_query_text)
    collection = CLIENT.get_or_create_collection(
        name="employee_manual_collection",
        embedding_function=None)
    results = collection.query(
        query_embeddings=user_query_embeddings,
        n_results=3,
        include= [
        "metadatas",
        "documents",
        "distances",
    ]
    )
    return results

# 根据检索结果 和用户的原始问题,组装到上下文(提示词),调用大模型, 生成最终的答案
def generate_answer(query_results, user_query_text):
    retrieval_chunks = query_results["documents"][0]
    content = "\n".join([f"--: {chunk}" for chunk in retrieval_chunks])
    prompt=f"""
    ##角色设定
    你是一个专业的人力资源专家,
    ## 任务描述
    根据用户的问题和公司的制度内容,回答问题
    ##输入数据
    用户的问题：{user_query_text}
    公司制度内容：{content}
    ##约束
    1:严格基于公司的制度内容回答问题,不要胡编乱造
    2:如果公司的制度内容没有相关信息,请明确说明“暂无此知识”
    3:回答简洁明了,条理清晰
    """
    completion = client.chat.completions.create(
        model="qwen3.7-plus",
        messages=[{"role": "user", "content": prompt}]
         )
    return completion.choices[0].message.content


if __name__ == '__main__':
    # all_text=load_file("D:\\python project\\H2602A\\boos-api\\llm\\employee_manual.docx")
    # chunks=split_text(all_text)
    # print(chunks)
    # embeddings = text_embedding(chunks)
    # print(embeddings)
    # save_to_vector_db(chunks)
    user_query_text= "员工上班时间超过多久算迟到"
    results = query_by_chroma(user_query_text)
    # print(results)
    final_answer = generate_answer(results, user_query_text)
    print(f"大模型生成的最终答案:{final_answer}")

读文档（load_file）与切片（split_text）：

load_file 用 python-docx 把段落纯文本拼成大字符串（未取表格）。 split_text 按 200 字符定长切片成列表。切片是为了让检索命中"小而准"的片段，粒度影响召回质量，入门用定值即可。

文本转向量：

同款 client，这次调 embedding 模型 text-embedding-v4 把文本变成向量（语义相近则向量相近，是检索基础）。函数把字符串自动包成列表，调用方不用管类型；再按每批 10 条分批请求，返回与输入顺序对应的向量列表。

向量库初始化与入库：

PersistentClient 持久化到本地磁盘，重启数据还在。 save_to_vector_db 拿到（或建）集合 employee_manual_collection，把每块对应的 ids / 原文 / 向量一起写入。建库是重操作，只跑一次。

检索与生成答案：

query_by_chroma 把问题转向量去库里召回最相关 3 段（带距离值可排序/过滤）。 generate_answer 取出片段拼进 prompt，约束模型"只基于制度内容答、没有就明说暂无此知识"——RAG 精髓就是每次拿实时召回的原文作答，有出处不瞎编。

总结

• 学历验证是 Function Calling：模型自己决定调工具、传参，结果回喂后总结，解决"操作外部系统"。

• RAG是检索增强：知识先向量化存库，提问召回片段喂模型，解决"基于私有知识准确作答"。

• 共同：走百炼地址、用 qwen 系列；都把脏活封装成独立函数、主流程只做编排。