本教程面向新手，全程分步讲解环境搭建、依赖安装、API Key 获取、代码配置及接口调用，所有代码可直接复制使用，确保零基础也能顺利完成项目部署与运行。

一、环境搭建：基础软件与虚拟环境配置

1.1 基础软件安装

需提前安装以下软件，确保版本适配：

• Python：推荐3.9-3.11版本（避免过高版本兼容问题），下载地址：https://www.python.org/downloads/，安装时勾选“Add Python to PATH”。

• Redis：用于缓存学历验证数据，下载地址：https://redis.io/download/，安装后启动服务（默认端口6379）。

• 代码编辑器：推荐 PyCharm（社区版即可）或 VS Code，方便代码编写与调试。

1.2 虚拟环境创建（推荐）

虚拟环境可避免依赖冲突，步骤如下：

1. 打开终端（CMD/PowerShell），进入项目目录（示例：D:\aizn\P4\2603-p4-boss-api-boss0721）。

2. 创建虚拟环境：python -m venv venv

3. 激活虚拟环境：

   1. Windows：venv\Scripts\activate

   2. Mac/Linux：source venv/bin/activate

4. 激活后，终端前缀会显示 (venv)，表示虚拟环境生效。

二、依赖安装：必备库及版本说明

本项目需安装以下依赖库，复制以下命令到激活的虚拟环境终端中执行：

pip install fastapi uvicorn openai==1.14.3 redis chromadb requests python-multipart

依赖说明：

• fastapi + uvicorn：构建 API 接口，启动服务；

• openai==1.14.3：适配阿里云灵析兼容接口，避免403权限错误（必须指定版本）；

• redis：操作 Redis 缓存；

• chromadb：构建向量知识库，用于员工手册查询；

• requests：发送 HTTP 请求，调用学历验证接口；

• python-multipart：处理接口请求参数（FastAPI 必备）。

三、关键配置：API Key 获取与环境变量设置

本项目需2个关键 API Key，分别用于阿里云灵析大模型调用、学历验证接口调用，以下是详细获取步骤。

3.1 阿里云灵析 API Key（DASHSCOPE_API_KEY）

1. 访问阿里云灵析官网：https://www.aliyun.com/product/dashscope，注册/登录阿里云账号。

2. 进入“灵析控制台”，点击左侧“API 密钥”，创建新的 API Key（选择“兼容 OpenAI 模式”）。

3. 复制生成的 API Key（格式：sk-xxxxxx），保存好（后续配置需用到）。

3.2 学历验证接口 API Key（MY_XXW_BGCX_API_KEY）

1. 访问学历验证接口官网：https://www.apimy.cn/，注册/登录账号。

2. 进入“用户控制台”（https://www.apimy.cn/user/key），找到“API 密钥”模块，申请免费密钥。

3. 复制生成的 Key，保存备用。

3.3 环境变量配置

在项目根目录下创建 .env 文件（无后缀），粘贴以下内容，替换为自己的 API Key：

DASHSCOPE_API_KEY=你的阿里云灵析API Key

MY_XXW_BGCX_API_KEY=你的学历验证接口API Key

注意：.env 文件需放在项目根目录，与 venv 文件夹同级。

四、代码实现：完整配置与核心逻辑说明

4.1 完整代码

import json
import os

import chromadb
import requests
from fastapi import APIRouter
from openai import OpenAI
from redis import Redis

from app.apis.llm.boss_rag.case7 import text_embedding
from app.models import Job
from app.schemas.llm_case1 import LlmCase3
from app.core.logging import logger

# 初始化路由
llm_day06_router = APIRouter(
    prefix="/llm-day06",
    tags=["LLM-DAY06"],
)

# Redis 客户端初始化（全局仅一次）
redis_client = Redis(
    port=6379,
    host="127.0.0.1",
    decode_responses=True,
    protocol=2
)

# 全局初始化 OpenAI 客户端（复用，避免重复实例化）
client = OpenAI(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="https://阿里云灵析兼容模式地址.cn-beijing.maas.aliyuncs.com/compatible-mode/v1",
)

# 全局初始化 Chroma 向量库（复用，避免频繁创建）
chroma_client = chromadb.PersistentClient(path=r"D:\aizn\P4\2603-p4-boss-api-boss0721\app\zhishi")
collection = chroma_client.get_collection(
    name="员工手册知识库集合",
    embedding_function=None
)

# 系统提示词
system_prompt = """
你是一个专业的智能助手，规则如下：
1. 当查询学历或者验证学历时，必须调用 学历验证工具 完成;
2. 当你想查询岗位相关信息时，必须调用 岗位查询工具 完成;
3. 当你想查询企业员工手册内容时，必须调用 知识库查询工具 完成;
"""

# 工具列表
tools = [
    {
        "type": "function",
        "function": {
            "name": "学历验证工具",
            "description": "当你想查询学历或者验证学历时非常有用。",
            "parameters": {
                "type": "object",
                "properties": {
                    "验证码": {
                        "type": "string",
                        "description": "学历验证码",
                    }
                },
                "required": ["验证码"]
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "岗位查询工具",
            "description": "当你想查询岗位相关信息时非常有用。",
            "parameters": {
                "type": "object",
                "properties": {
                    "岗位ID": {
                        "type": "string",
                        "description": "岗位ID",
                    }
                },
                "required": ["岗位ID"]
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "知识库查询工具",
            "description": "当你想查询企业员工手册内容时非常有用。",
            "parameters": {
                "type": "object",
                "properties": {
                    "用户问题": {
                        "type": "string",
                        "description": "用户的问题",
                    }
                },
                "required": ["用户问题"]
            },
        },
    }
]

# 工具1：验证学历
def 学历验证工具(arguments):
    if arguments is None:
        return "请提供正确的学历验证码"
    验证码 = arguments["验证码"]
    if 验证码 is None:
        return "请提供正确的学历验证码"
    key = f"boss:llm:academic_credential_verification:{验证码}"
    redis_verification_data = redis_client.get(key)
    if redis_verification_data is None:
        BASE_URL = "https://www.apimy.cn/api/xxw/bgcx"
        payload = {"key": os.getenv("MY_XXW_BGCX_API_KEY"), "vcode": arguments["验证码"]}
        headers = {"Content-Type": "application/json"}
        try:
            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)
        except Exception as e:
            return f"学历验证失败：{str(e)}"
    else:
        return redis_verification_data

# 工具2：查询岗位
async def 岗位查询工具(arguments):
    if arguments is None:
        return "请提供正确的岗位查询参数"
    岗位ID = arguments["岗位ID"]
    print(f"岗位ID:{岗位ID}")
    job = await Job.get_or_none(id=岗位ID)
    if job is None:
        return "没有此岗位"
    job_info = f"职位名称:{job.job_name}"
    return job_info

# 工具3：查询知识库（复用全局客户端）
def 知识库查询工具(arguments):
    if arguments is None:
        return "请提供正确的知识库查询参数"
    用户问题 = arguments["用户问题"]
    try:
        用户问题_嵌入向量 = text_embedding(用户问题)
        results = collection.query(
            query_embeddings=用户问题_嵌入向量,
            n_results=5,
            include=["metadatas", "documents", "distances"]
        )
        检索内容块 = results["documents"][0]
        内容 = "\n".join([f"--: {chunk}" for chunk in 检索内容块])
    except Exception as e:
        return f"知识库查询失败：{str(e)}"

    prompt = f"""
## 角色设定
你是一个专业的人力资源专家,
## 任务描述
根据用户的问题和公司的制度内容,回答问题
## 输入数据
用户的问题:{用户问题}
公司制度内容:{内容}
## 约束
1:严格基于公司的制度内容回答问题,不要胡编乱造
2:如果公司的制度内容没有相关信息,请明确说明“暂无此知识”
3:回答简洁明了,条理清晰
"""
    print(f"prompt:{prompt}")
    try:
        completion = client.chat.completions.create(
            model="qwen-plus",
            messages=[{"role": "user", "content": prompt}],
            temperature=0.75,
        )
        print(completion.model_dump_json())
        return completion.choices[0].message.content
    except Exception as e:
        return f"大模型调用失败：{str(e)}"

# 获取 AI 响应
def get_ai_response(messages):
    try:
        completion = client.chat.completions.create(
            model="qwen-plus",
            messages=messages,
            temperature=0.75,
            tools=tools
        )
        return completion
    except Exception as e:
        raise Exception(f"AI 响应获取失败：{str(e)}")

# 接口：LLM 对话
@llm_day06_router.post("/llm_chat")
async def llm_chat(request: LlmCase3):
    # 每个请求独立会话，避免全局变量污染
    messages = [{"role": "system", "content": system_prompt}]
    user_message = {"role": "user", "content": request.user_query}
    messages.append(user_message)

    MAX_LOOP = 5  # 限制工具调用轮次，防止死循环
    for _ in range(MAX_LOOP):
        try:
            completion = get_ai_response(messages)
            msg = completion.choices[0].message
            messages.append(msg)
            # 无工具调用，直接返回结果
            if msg.tool_calls is None:
                return {
                    "code": 1,
                    "data": msg.content
                }
            # 执行工具调用
            func_mapping = {
                "知识库查询工具": 知识库查询工具,
                "岗位查询工具": 岗位查询工具,
                "学历验证工具": 学历验证工具,
            }
            for tool in msg.tool_calls:
                tool_id = tool.id
                func_name = tool.function.name
                func_arguments = json.loads(tool.function.arguments)
                tool_result = func_mapping[func_name](func_arguments)
                # 异步工具需 await
                if hasattr(tool_result, "__await__"):
                    tool_result = await tool_result
                tool_message = {
                    "role": "tool",
                    "content": tool_result,
                    "tool_call_id": tool_id
                }
                messages.append(tool_message)
        except Exception as e:
            return {"code": -1, "data": f"接口调用失败：{str(e)}"}

    return {"code": -1, "data": "工具调用超过最大轮次，无法获取结果"}

4.2 关键配置说明

• 向量库路径：chromadb.PersistentClient(path=r"D:\aizn\P4\2603-p4-boss-api-boss0721\app\zhishi")，需确保该路径存在，且已导入员工手册相关文档（可通过 Chroma 官方工具导入）。

• Redis 配置：默认本地 Redis 服务，若 Redis 部署在其他服务器，需修改 host 为对应服务器 IP。

• OpenAI 客户端：复用全局实例，避免重复初始化导致的403权限错误。

五、服务启动与接口测试：实操步骤

5.1 启动 Redis 服务

打开新终端，启动 Redis：redis-server（若 Redis 已配置为服务，可直接在服务中启动）。

5.2 启动 FastAPI 服务

1. 回到激活的虚拟环境终端，进入项目根目录。

2. 启动服务：uvicorn app.main:app --reload（假设项目主文件为 main.py，需根据实际项目结构调整）。

3. 启动成功后，终端会显示服务地址：http://127.0.0.1:8000。

5.3 接口调用测试

可通过 FastAPI 自带的 Swagger 文档测试接口，步骤如下：

1. 访问 Swagger 地址：http://127.0.0.1:8000/docs。

2. 找到接口，点击“Try it out”。

3. 输入请求参数（示例）：

   1. 查询学历：{"user_query": "验证学历，验证码为123456"}

   2. 查询岗位：{"user_query": "查询岗位ID为1的岗位信息"}

   3. 查询员工手册：{"user_query": "员工试用期规定是什么"}

4. 点击“Execute”，查看接口返回结果，验证是否正常调用。

六、常见问题排查：快速定位与解决

6.1 403权限错误（Workspace endpoint access denied）

原因：openai 库版本过高，需降级到1.14.3。解决方法：执行 pip install openai==1.14.3，重启服务（需完全终止进程，热重载无效）。

6.2 学历验证接口403错误

原因：API Key 不正确或未生效。解决方法：检查 .env 中 MY_XXW_BGCX_API_KEY 是否正确，重新申请密钥并配置。

6.3 知识库查询无结果

原因：向量库未导入相关文档，或查询参数错误。解决方法：确认 Chroma 向量库路径正确，已导入员工手册文档，检查 text_embedding 函数是否正常工作。

6.4 Redis 连接失败

原因：Redis 服务未启动，或端口/IP 配置错误。解决方法：启动 Redis 服务，检查代码中 Redis 配置是否与实际一致。

七、新手总结：核心要点回顾

本项目核心是通过 FastAPI 构建接口，结合 Redis、Chroma 及大模型工具，实现学历验证、岗位查询、员工手册查询三大功能。新手需重点掌握：虚拟环境使用、API Key 配置、依赖库版本控制、接口测试方法。遇到问题优先检查环境配置、依赖版本及 API Key 有效性，逐步排查即可解决。