10.14面试高级python和agent- 惊觉

1. Python GIL 是什么？项目中如何使用的

回答

GIL (Global Interpreter Lock)，全局解释器锁

CPython 特有的锁，同一时刻一个 Python 进程内只能有一个线程执行 Python 字节码。
GIL 不会锁住 IO，IO 阻塞的时候 GIL 会释放；但是 CPU 密集型计算，多线程无法利用多核。
两个结论：
1. IO 密集型任务（网络请求：调用 LLM、Milvus、Redis、HTTP 接口）：多线程 /asyncio 异步可以获得收益，等待网络的时候释放 GIL。
2. CPU 密集（大量文本分片、Embedding 计算）：多线程没用，要用多进程，绕过 GIL。

✅ 项目场景（你的 Agent 项目）

FastAPI 里的接口请求，属于IO 密集：调用 LLM、Milvus、Redis、数据库，大量时间在等待网络 IO，所以我们用异步 async。IO 等待期间释放 GIL，并发处理更多请求。
文档流水线：文档分片、文本清洗、Embedding 推理是 CPU 密集，这部分单独开多进程池处理，绕开 GIL，不占用 web 服务主进程。

坑点：不要在 FastAPI 同步接口里跑大量 CPU 计算，会阻塞整个 worker。

代码示例（多进程处理文档分片，规避 GIL）

from multiprocessing import Pool

def split_document(text: str):
    """CPU密集：文档分片，放到子进程执行，绕开GIL"""
    chunks = []
    # 标题感知分片逻辑
    return chunks

if __name__ == "__main__":
    texts = ["文档文本1", "文档文本2"]
    with Pool(processes=4) as pool:
        result = pool.map(split_document, texts)

口述一句话总结：

GIL 是 CPython 全局解释锁，限制单进程同一时刻只有一个线程跑 Python 字节码。我们 Agent 服务，IO 密集的 LLM/Milvus 网络请求用异步；CPU 密集的文档预处理、embedding 向量化，用多进程绕过 GIL。

2. Python 中间件知道些什么，项目怎么使用

回答

Python 中间件，在 FastAPI 里就是请求 / 响应的拦截钩子，在路由处理前后执行通用逻辑。
FastAPI 中间件基于Starlette中间件体系。
常见中间件类型：

内置：CORSMiddleware（跨域）
自定义中间件：全链路 traceId 注入、日志埋点、请求耗时统计、全局异常捕获、鉴权校验。

✅ 项目场景（你的 Agent 项目）
我们自定义了一个Trace 中间件：

请求进来生成 traceId，注入 request.state，向下透传给所有下游调用（Milvus、Redis、LLM 请求）
记录请求开始时间，接口返回时计算耗时，打印结构化日志
统一捕获全局未处理异常，包装成统一错误返回体

注意：中间件分同步中间件和@asynccontextmanager异步中间件，异步项目必须用异步中间件。

代码示例

from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
import uuid
import time

class TraceMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # 请求进入：生成traceId
        trace_id = str(uuid.uuid4())
        request.state.trace_id = trace_id
        start_time = time.time()
        
        response = await call_next(request)
        
        # 返回前：记录耗时
        cost = time.time() - start_time
        response.headers["X-Trace-Id"] = trace_id
        print(f"trace={trace_id}, path={request.url.path}, cost={cost}")
        return response

# 注册到app
# app.add_middleware(TraceMiddleware)
# app.add_middleware(CORSMiddleware, allow_origins=["*"], ...)

口述总结：
FastAPI 基于 Starlette 中间件，项目中用到 CORS 内置中间件，自定义 Trace 中间件做全链路追踪、耗时埋点，统一日志，traceId 贯穿 Agent 整个 RAG 和 LLM 调用链路。

3. FastAPI 工具函数有使用嘛，项目中使用场景

回答

FastAPI 工具函数，一般指Depends依赖注入函数（最核心），还有 HTTPException、BackgroundTasks 等。

依赖注入 Depends：抽取可复用逻辑，自动注入到路由函数。

✅ 项目场景 Agent 项目

依赖函数get_tenant_context：从 request.state 读取 token 解析出来的 tenantId，校验租户状态，注入到 Agent 接口，所有 RAG / 工具调用路由复用。
数据库会话依赖：get_db，获取 PG 连接会话。
BackgroundTasks 后台任务：异步写入审计日志，不阻塞主请求（Agent 调用 LLM 成功后，把日志丢后台任务入库）。

代码示例

from fastapi import Depends, BackgroundTasks, Request, HTTPException
from pydantic import BaseModel

async def get_tenant_ctx(request: Request):
    """依赖函数：获取租户上下文，所有Agent接口复用"""
    tenant_id = request.state.tenant_id
    if not tenant_id:
        raise HTTPException(status_code=401, detail="tenant invalid")
    return tenant_id

@app.post("/agent/chat")
async def agent_chat(
    req: BaseModel,
    tenant_id: str = Depends(get_tenant_ctx),
    bg: BackgroundTasks = BackgroundTasks()
):
    result = await run_agent(req.query, tenant_id)
    # 后台任务，异步写审计日志，不阻塞主链路
    bg.add_task(save_audit_log, tenant_id, result)
    return result

口述总结：大量使用 Depends 依赖函数抽取租户上下文、数据库会话；BackgroundTasks 后台任务做审计日志异步落库，不阻塞 Agent 主请求链路。

4. FastAPI 异步和同步使用场景；Agent 服务应该异步还是同步

回答

区分场景

async def 异步接口：适合 IO 密集任务。等待网络（LLM、Milvus、Redis、DB），await 释放事件循环，worker 可以处理其他请求。
适用：Agent 对话接口、RAG 检索、调用 LLM、向量库查询。

✅ Agent 主链路推荐异步，Agent 核心链路全部是 IO 等待：LLM 调用、Milvus 检索、Redis 会话读取、PostgreSQL 查询，大量 IO 等待，异步可以提升并发。

def 同步接口：适合 CPU 密集任务，或者调用同步第三方 SDK（部分老 SDK 不支持 async）。
同步接口 FastAPI 会自动丢到线程池执行；不适合大量并发 IO 场景，线程池上限有限。

注意坑：异步函数内部不能直接调用同步阻塞 SDK（同步 Redis、同步 Milvus 客户端），会卡死事件循环；要用 async 版本 SDK，或者用asyncio.to_thread()包装同步阻塞调用。

代码示例（同步 SDK 放到线程池执行，不阻塞事件循环）

import asyncio
import milvus # 假设milvus是同步SDK

async def rag_search(tenant_id:str, query:str):
    # 同步Milvus调用，放到线程池
    res = await asyncio.to_thread(milvus.search, tenant_id, query)
    return res

口述总结：Agent 服务主链路全部用 async 异步接口，因为大量 IO 等待（LLM、向量库、Redis），提高并发；如果调用老的同步 SDK，使用 asyncio.to_thread 丢入线程池，防止阻塞事件循环；纯 CPU 密集的文档预处理，放到独立进程，不放在 web 接口里。

5. FastAPI 迭代器和生成器区别、使用场景

面试官实际想问 Python 原生迭代器 / 生成器，在 FastAPI SSE / 流式输出场景（Agent 大模型流式返回）。

回答

迭代器 (Iterator)：实现__iter__ + __next__，保存状态，逐个返回元素，手动迭代。
生成器 (Generator)：用yield关键字，是迭代器的一种简化写法，函数执行到 yield 暂停，下次迭代从断点继续，自动维护上下文状态，内存友好，不用一次性把全部数据加载到内存。

✅ 项目场景：Agent SSE 流式输出（重点！）
SSE 流式返回 LLM 的 token 流，用异步生成器 async def + yield，一段一段吐出 LLM 返回的内容，不需要等大模型全部生成完一次性返回，前端实时展示文字。

代码示例：异步生成器用于 SSE

import asyncio

async def llm_stream_generator(query: str):
    """异步生成器，每次yield一段token"""
    for chunk in await call_llm_stream(query):
        yield f"data: {chunk}\n\n"
        await asyncio.sleep(0.01)

@app.get("/agent/stream")
async def stream_chat():
    from fastapi.responses import StreamingResponse
    return StreamingResponse(llm_stream_generator("用户提问"), media_type="text/event-stream")

口述总结：生成器是迭代器的简化实现，基于 yield。项目中 SSE 流式输出 LLM token，使用异步生成器，分段返回数据，不需要把全部回答加载进内存，适合 Agent 实时对话场景。

6. FastAPI 函数装饰器有哪些，怎么用

回答

FastAPI 常用装饰器：

@app.get / @app.post / @app.put：注册路由，最基础
@app.websocket：websocket 路由
自定义装饰器：用来做接口拦截、权限校验、日志统计（注意：异步接口要写异步装饰器）

补充：Python 原生装饰器原理：包装函数，在函数执行前后插入逻辑。

代码示例：自定义装饰器，统计接口耗时

import time
from functools import wraps
from fastapi import Request

def api_timer(func):
    @wraps(func)
    async def wrapper(request: Request, *args, **kwargs):
        start = time.time()
        res = await func(request, *args, **kwargs)
        print(f"route cost: {time.time() - start}")
        return res
    return wrapper

@app.post("/agent/chat")
@api_timer # 自定义装饰器
async def chat(request: Request):
    return {"msg":"ok"}

口述总结：内置@app.get/post路由装饰器；项目写自定义异步装饰器统计接口耗时。但大部分通用逻辑，优先用中间件 / Depends 依赖，而不是装饰器。因为装饰器不好统一管理，新增路由容易忘记加。

7. 解析式下python中@的用法？

回答

@是一个装饰器，针对函数，起调用传参的作用。有修饰和被修饰的区别，‘@function'作为一个装饰器，用来修饰紧跟着的函数（可以是另一个装饰器，也可以是函数定义）。

复制代码
def funA(desA):
 print("It's funA")
 
def funB(desB):
 print("It's funB")
 
@funB
@funA
def funC():
 print("It's funC")
复制代码

比如上面输出的是

It's funA 
It's funB

@funB 修饰装饰器@funA，@funA 修饰函数定义def funC()，将funC()赋值给funA()的形参，再将funA(funC())赋值给funB()。 执行的时候由上而下，先定义funA、funB，然后运行funB(funA(funC()))。
此时desA=funC()，然后funA()输出‘It’s funA’；desB=funA(funC())，然后funB()输出‘It’s funB’。

8. Embedding 用什么模型，如何使用

回答

贴合项目：私有化 / API 调用 Embedding 模型，比如 bge-large-zh，或者厂商 embedding 接口。

模型：BGE-Large-zh，中文效果好，适合金融 / 知识库文档向量化。
使用流程：

文档分片后，把 chunk 文本送入 Embedding 模型，输出 float 向量数组。
入库 Milvus；用户 query 也用同一个 Embedding 模型生成向量，做向量检索。

重点：Query 和文档必须用同一个 Embedding 模型，否则向量空间不一致，召回失效。

代码示例

from sentence_transformers import SentenceTransformer

# 加载bge-large-zh
model = SentenceTransformer("BAAI/bge-large-zh")

def get_embedding(text:str):
    vec = model.encode(text, normalize_embeddings=True)
    return vec.tolist()

chunk_text = "监管规则文档片段"
vector = get_embedding(chunk_text)
# 写入Milvus
# milvus.insert(vector=vector, chunk_text=chunk_text, tenant_id="xxx")

口述：项目使用 BGE-Large-zh 作为 Embedding 模型，文档分片和用户 query 都用同一个模型生成向量；向量做归一化，存入 Milvus；检索时用户问题生成向量做相似度匹配。

可选：如果是调用线上 API，就换成 http 请求调用 embedding 接口。

9. Agent Skills/Tools/MCP 区别，项目用过哪些，代码解释如何调用?

回答

概念区分

Tool（工具 / Function Call）
LLM 原生能力：LLM 识别意图，输出结构化函数调用参数，由程序执行函数，把结果丢回给 LLM 继续推理。
本质：LLM 驱动的函数调用，OpenAI function call / Anthropic tool call 都属于这一类。

粒度：单个可调用函数，比如查询工单、计算风险分。

Skill（技能）
一组 Tool + 执行逻辑 + 提示词 + 业务编排，封装成一个完整能力。
Skill 是更高一层的业务抽象，一个 Skill 可以包含多个 Tool、多轮工具调用、前置校验、结果后处理。

举例：【工单风险评估 Skill】，内部会连续调用query_work_order、risk_score_calculate两个 tool，并且内置业务判断逻辑，这整套打包叫 Skill。

简单一句话区分：

Tool = 最小可调用函数单元
Skill = 业务能力包，由多个 Tool + 编排逻辑组成

MCP（Model Context Protocol）
MCP 是模型上下文协议，标准化协议，用来让 Agent 安全、标准化调用外部服务 / 资源（文件、数据库、API、第三方系统）。

和普通 Function Call 最大差别：
普通 Tool：工具定义、调用逻辑、参数解析全部写在你自己 Agent 代码里面，自定义实现，没有统一标准。
MCP：遵循 MCP 协议，Agent 作为客户端，连接 MCP Server；MCP Server 向外暴露可调用工具列表，工具能力由外部 MCP 服务提供，Agent 不需要写每个工具的底层实现。
优势：工具可插拔，新增外部能力不用改 Agent 核心代码；支持权限、沙箱隔离。
局限：项目中如果是内部自研 Agent，一般不会上 MCP，MCP 适合对接大量异构外部系统。

✅ 你的项目场景（金融质检 Agent）

没有引入 MCP（可以直接说：项目当时没有接入 MCP，MCP 属于较新的标准，我们自研工具适配器做函数调用）
一共封装 3 个 Tool（最小函数单元）
1. query_work_order：查询工单详情
2. risk_score_calculate：风险打分
3. update_task_status：更新质检任务状态
封装 1 个业务 Skill：quality_risk_audit_skill（质检风险评估 Skill），内部编排调用上面 3 个 Tool，加前置参数校验、结果格式化。

一句话口述总结：
Tool 是最小的可调用函数；Skill 是把多个 Tool 加上业务逻辑封装成一个完整业务能力；MCP 是标准化协议，用来让 Agent 对接外部服务的工具，我们项目自研 Agent 没有接入 MCP，自己实现 Tool 适配器。

步骤 1：定义工具描述（给 LLM 识别用的 JSON Schema）

from pydantic import BaseModel, Field
import json

# ========== Tool定义（最小调用单元） ==========
class QueryWorkOrderReq(BaseModel):
    order_id: str = Field(description="工单唯一ID")

def query_work_order(order_id: str):
    """Tool：查询工单信息，同步调用业务服务"""
    # 调用业务微服务http接口
    # resp = requests.get(f"http://biz-api/order/{order_id}")
    return {"order_id": order_id, "amount": 10000, "create_time": "2026-09-01"}

def risk_score_calculate(order_data: dict):
    """Tool：风险打分"""
    score = 0
    if order_data["amount"] > 5000:
        score = 70
    return {"risk_score": score}

def update_task_status(task_id: str, status: str):
    """Tool：更新任务状态，写操作"""
    return {"task_id": task_id, "new_status": status}

# 工具元信息，传给LLM，用于function call识别
tool_list = [
    {
        "type": "function",
        "function": {
            "name": "query_work_order",
            "description": "查询工单详情，输入工单ID，返回工单基础信息",
            "parameters": QueryWorkOrderReq.model_json_schema()
        }
    },
    {
        "type": "function",
        "function": {
            "name": "risk_score_calculate",
            "description": "根据工单数据计算风险分数",
            "parameters": {
                "type": "object",
                "properties": {"order_data": {"type": "object"}}
            }
        }
    },
    {
        "type": "function",
        "name": "update_task_status",
        "description": "更新质检任务状态",
        "parameters": {
            "type": "object",
            "properties": {
                "task_id": {"type": "string"},
                "status": {"type": "string"}
            },
            "required": ["task_id", "status"]
        }
    }
]

步骤 2：Tool 适配器（统一分发，Agent 核心引擎不感知底层实现，就是你前面架构讲的 ToolAdapter）

# Tool适配器：统一路由，根据LLM返回的工具名称，执行对应函数
class ToolAdapter:
    def __init__(self):
        self.tool_map = {
            "query_work_order": query_work_order,
            "risk_score_calculate": risk_score_calculate,
            "update_task_status": update_task_status
        }

    async def run_tool(self, tool_name: str, tool_args: dict):
        """执行工具，统一捕获异常，标准化返回"""
        if tool_name not in self.tool_map:
            return {"error": f"不存在工具{tool_name}"}
        func = self.tool_map[tool_name]
        # 如果是同步函数，放到线程池，不阻塞async事件循环
        result = await asyncio.to_thread(func,** tool_args)
        return result

步骤 3：Skill 封装：质检风险评估 Skill（多个 Tool 编排）

# Skill：业务能力封装，编排多轮tool调用
class QualityRiskAuditSkill:
    def __init__(self, adapter: ToolAdapter):
        self.adapter = adapter
        self.max_round = 3 # 限制最大tool调用轮次，防止死循环

    async def run(self, task_id: str, order_id: str):
        """Skill主流程：一整套业务编排"""
        # Step1：调用tool查询工单
        order_info = await self.adapter.run_tool("query_work_order", {"order_id": order_id})
        if "error" in order_info:
            return order_info
        # Step2：调用tool计算风险分
        risk_res = await self.adapter.run_tool("risk_score_calculate", {"order_data": order_info})
        # Step3：调用tool更新任务状态
        await self.adapter.run_tool("update_task_status", {"task_id": task_id, "status": "ai_audited"})
        return {
            "order_info": order_info,
            "risk_result": risk_res
        }

步骤 4：Agent 主逻辑：LLM 判断要不要调用 Tool/Skill

async def agent_run(user_query: str, session_id: str):
    adapter = ToolAdapter()
    audit_skill = QualityRiskAuditSkill(adapter)

    # 1. 调用LLM，传入工具列表，让LLM判断意图
    llm_resp = await call_llm(messages=[{"role":"user","content":user_query}], tools=tool_list)
    
    # 2. LLM返回工具调用指令
    if llm_resp.get("tool_calls"):
        for tool_call in llm_resp["tool_calls"]:
            t_name = tool_call["function"]["name"]
            t_args = json.loads(tool_call["function"]["arguments"])
            # 执行tool
            tool_result = await adapter.run_tool(t_name, t_args)
            # 把工具执行结果塞回消息列表，再次交给LLM做推理
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call["id"],
                "content": json.dumps(tool_result)
            })
        # 二次调用LLM，结合工具返回结果生成最终回答
        final_resp = await call_llm(messages)
        return final_resp["content"]
    else:
        # 不需要调用工具，直接返回LLM回答
        return llm_resp["content"]

面试官高频追问（配套回答）

Q1：什么时候用 Tool，什么时候封装成 Skill？

回答：
Tool 是最小单元，单一原子操作，只做一件事；
如果业务需要固定顺序连续调用多个 Tool，带前置校验、后置结果组装、固定业务规则，就封装成 Skill。
比如风险评估必须：查工单→算分→更新状态，固定流程，封装成 Skill，Agent 直接调用这个 Skill，不用 LLM 反复决定每一步工具。

Q2：MCP 和你这个自研 Tool 适配器的差异？

回答：
自研 Tool 适配器：工具定义、执行代码全部在项目内部，适合内部业务 API，简单可控。缺点：新增工具需要改代码、重新发布服务。
MCP 是标准协议，Agent 作为客户端去连接独立 MCP Server，MCP 服务向外暴露工具。新增工具只需要在 MCP 服务配置，不需要修改 Agent 核心代码，适合对接外部异构系统、文件、数据库，做插件化 Agent。
我们项目是内部业务质检，工具数量少，迭代可控，所以自研适配器，没有引入 MCP。

Q3：Tool 调用死循环怎么避免？

回答：
两层防护：

全局设置最大工具调用轮次，超过直接终止；
每次拿到工具返回结果，Prompt 引导 LLM 判断信息是否足够回答用户，不需要继续调用工具。

Q4：Tool 返回结果很大怎么办？

回答：
对 tool 返回结果做摘要、截断，只保留 LLM 推理需要的关键字段，减少 token 消耗，避免 prompt 超长。

Q5：Tool 调用存在并发 / 数据脏写问题怎么处理？

回答：
写类型 tool（update_task_status）增加幂等 key，防止 LLM 重复触发工具调用，重复更新状态。

10. Agent 幻觉如何减少，如何提高准确性，举例场景

回答

幻觉本质：LLM 在没有足够参考信息时编造内容。多层策略组合：

RAG 层：分片优化（标题感知分片）、多路召回 + 重排，提升参考文档质量，把正确的上下文给到 LLM。
Prompt 约束：强制要求模型只能基于提供的参考文档回答；无相关信息直接回答不知道，禁止编造；输出附带引用 chunk_id 溯源。
结果校验层：对输出结果做校验，引用分片相似度低于阈值直接拒绝生成答案。
业务校验：金融质检场景，AI 输出仅作为辅助，人工复核。
拒绝编造策略：召回分数低于阈值，直接返回知识库无相关信息，不送入 LLM。

✅ 场景例子（金融质检）

当 Agent 判断录音是否违规，必须基于召回的监管文档片段；Prompt 强制要求：如果参考文档没有对应的监管条款，不能自行编造违规结论，同时输出引用的分片 ID 用于溯源。

代码示例（Prompt 约束模板）

system_prompt = """
你只能基于【参考文档】回答用户问题。
如果参考文档没有对应信息，直接回复“没有找到相关依据”，禁止编造任何规则。
输出结果附带引用的分片ID。
参考文档：{context}
"""

口述总结：幻觉是组合方案，RAG 提升召回质量，Prompt 强约束，低相似度召回直接拦截，输出强制溯源；金融场景 AI 结果还要人工复核。

11. Agent 多轮对话怎么设计

回答

整体思路：会话 ID 隔离，上下文分段存储，长短记忆分离

生成唯一 session_id，绑定租户 + 用户，作为会话唯一标识。
短期记忆（Redis）：存储最近多轮对话（user、assistant、tool 调用记录），设置 TTL；每次请求读取历史，拼入 prompt。
上下文窗口管理：对话累积过长时，做摘要压缩，防止 token 超限。
长期记忆（PostgreSQL）：会话全量记录异步持久化，用于审计，不会全部塞到 prompt。
会话隔离：不同 session、不同租户完全隔离。

存储结构：Redis key session:{tenant_id}:{session_id}，value 是 json 数组，每一条包含 role, content, tool_calls。

示例伪代码

# 读取历史
history = await redis.get(f"session:{tenant_id}:{session_id}")
messages = json.loads(history)
messages.append({"role":"user", "content":user_query})
# 调用LLM
resp = await llm.chat(messages=messages)
messages.append({"role":"assistant", "content":resp})
# 写回Redis
await redis.set(f"session:{tenant_id}:{session_id}", json.dumps(messages), ex=3600)

口述总结：多轮对话使用 sessionId 隔离，短期对话放 Redis，带过期时间；每次请求读取历史拼接消息；上下文过长自动摘要压缩；会话结束异步写入 PG 审计。

12. Agent LLM 超时如何处理，如何优化

回答

处理策略（分层）

请求层超时设置：LLM http 客户端设置 timeout，区分连接超时、读取超时。
重试策略：网络抖动导致的短暂超时，做有限次数重试（指数退避；业务敏感场景不能无限重试，防止重复调用 LLM）。
熔断降级：滑动窗口统计 LLM 接口失败率，连续失败达到阈值，触发熔断，不再调用 LLM，直接返回降级结果。
链路超时控制：FastAPI、Nginx 设置最大超时，防止请求长时间挂住占用 worker。

优化手段

减少单次 prompt token 数量：上下文摘要、缩短参考分片。
开启 LLM 流式 SSE，提前返回 token，前端逐步渲染，用户感知等待变短。
缓存高频相同 query 结果，避免重复调用 LLM。

代码示例：aiohttp 客户端超时 + 重试

import aiohttp
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(2), wait=wait_exponential(multiplier=1, min=1, max=3))
async def call_llm(prompt:str):
    timeout = aiohttp.ClientTimeout(total=15) # 总超时15s
    async with aiohttp.ClientSession(timeout=timeout) as session:
        resp = await session.post("llm-api", json={"prompt":prompt})
        return await resp.json()

口述总结：设置客户端超时；网络抖动做有限次指数退避重试；熔断机制，LLM 大面积故障直接降级；优化上做上下文压缩、SSE 流式输出、高频问答缓存。

13. Agent SSE 结构是什么，项目怎么使用；如果存储 SSE 流到 Redis，Redis 用什么结构

回答

SSE (Server-Sent Events) 结构

SSE 是服务端单向推送的 http 协议，基于长连接；响应头Content-Type: text/event-stream。
消息格式：

data: 消息内容\n\n

可以带 event、id 字段。适合 LLM 流式输出，一段一段推送 token 给前端。

✅项目使用：Agent 对话接口，StreamingResponse + 异步生成器，实时返回 LLM 流式 token，前端打字机效果。

如果要把 SSE 流存入 Redis：选 List（列表）结构

原因：SSE 是有序的流式分段，一段一段按顺序产生。

key：sse_stream:{session_id}
每次产生一个 chunk，LPUSH / RPUSH追加到 list。
读取按顺序 range 获取所有片段，重建完整回答。

不适合 String：流式分段不断追加，字符串会不断扩容，性能差。
不适合 Hash：hash 适合 k-v 键值对，不适合有序流。

代码示例

# 写入每一段sse chunk到redis list
await redis.rpush(f"sse_stream:{session_id}", json.dumps({"chunk":token_chunk}))
# 读取全部片段
chunks = await redis.lrange(f"sse_stream:{session_id}", 0, -1)

口述总结：SSE 是 http 长连接，服务端单向推送，固定 data:xxx 格式；项目用 StreamingResponse + 异步生成器实现 LLM 打字机流式返回。保存 SSE 分段流到 Redis，选用 List 有序列表，RPUSH 追加每一段 token 分片。

追加高频面试题（Python + FastAPI + Agent/RAG，继续准备）

14. FastAPI 的 Depends 和 BackgroundTasks 区别？

Depends：请求执行前 / 中，注入依赖，同步 / 异步，用来获取上下文、db 会话。
BackgroundTasks：响应返回给前端之后，后台执行任务，不阻塞主请求。
项目场景：BackgroundTasks 用来异步写审计日志，Depends 用来获取租户上下文。

15. 异步代码里面如果写 time.sleep () 会发生什么？怎么替换？

time.sleep () 是阻塞调用，会阻塞整个事件循环。异步代码必须用await asyncio.sleep()。

16. 什么是 FastAPI Pydantic，项目怎么用？

Pydantic 做请求体、参数校验，自动类型转换。Agent 项目：入参校验、Function-Call 参数 schema 定义，LLM 输出结构化解析。

from pydantic import BaseModel
class AgentChatReq(BaseModel):
    query: str
    session_id: str

17. RAG 多路召回是什么，BM25 和向量检索优缺点

向量召回：语义匹配；对字面不一样、语义相同的问题召回好；缺点：依赖 Embedding，计算量大。
BM25 关键词检索：字面关键词匹配；对专有名词效果好；无法理解语义。
多路召回：两个结果拿到，重排序融合，取 topN，互补。

18. LLM 流式输出 SSE 和 websocket 区别？

SSE：单向，http 长连接，服务推送给前端，简单，自动重连；适合问答打字机。
Websocket：双向全双工，前后端互相发消息；适合聊天双向实时交互，协议复杂。
项目 Agent 流式回答选用 SSE 足够。

19. Redis 的过期策略是什么？如果大量 session 过期，会发生什么

Redis 过期：惰性删除 + 定期删除。大量 key 同时过期会造成 CPU 抖动。
项目规避：session TTL 加随机偏移量，避免大量会话同时过期。

20. Milvus 索引类型，什么时候选什么索引

IVF_FLAT：小数据集，高精度；IVF_SQ8：压缩向量，节省内存，大数据量，精度轻微下降。
项目知识库使用 IVF_SQ8，平衡内存和检索速度。

21. 什么是 Agent 的 Tool 调用循环终止条件？

设置最大调用轮次（例如最多 3 轮 function call）
检测工具返回结果已经足够回答用户问题，不再继续调用工具
防止无限循环调用 tool。

22. FastAPI 的 worker 类型：uvicorn 的 uvloop 和 workers，同步 / 异步 worker 选型

uvicorn 默认用 uvloop，提升异步 IO 性能；异步代码用单个 worker+uvloop，多 worker 会多进程。

注意：多进程会复制内存，Embedding 模型加载到多进程会占用大量内存。项目中 Embedding 服务独立部署，不和 web 服务放一起。

23. 什么是 Starlette，和 FastAPI 关系

FastAPI 基于 Starlette（轻量异步 web 框架），Starlette 负责底层 http、中间件、StreamingResponse；FastAPI 在上面增加 Pydantic、路由、Depends 等能力。

24. 异步代码的异常捕获怎么处理？

使用 try/await，在异步函数内捕获异常；全局中间件捕获未处理异常，包装为统一返回。

这一组是 Agent/RAG 工程面试题

背景：RAG 检索上下文、文本切片、意图识别，属于 Agent 知识库问答里最容易踩坑的地方，面试官重点想听：你怎么权衡、怎么做验证、怎么判断好坏，而不是只说固定数值

1. 上下文长度选择：链接多长、条数取舍（RAG 检索 TopN）

核心问题：检索返回多少条文档片段放进 prompt，TopN 怎么定？

思路

不是固定写 Top3/Top5，是预算优先：LLM 上下文窗口剩余 token 预算 → 单条切片平均 token → 反推最大条数

先算可用 token：可用上下文token = 模型总窗口 - 系统prompt - 用户query - 输出预留token

例：模型 8k 窗口，系统 prompt 占 1k，预留输出 1k → 可用 6k 给检索片段。如果切片每条 800token，最多放 7 条。

排序策略：按向量相似度 + 重排（Rerank）分数，不是只靠向量距离。
取舍规则（取舍逻辑）
- 优先保留：高 rerank 分、和 query 语义强相关；时间匹配（比如客户故障文档取最新）；来源权威（官方 runbook > 论坛笔记）
- 直接丢弃：低相似度、内容重复、和问题无关；超长冗余段落
- 兜底：达到 token 上限后，截断低分数片段，优先保证高质量片段完整，不要粗暴对半切
工程调优策略
- 动态 TopN：不是固定条数。简单意图少拿片段，复杂故障 / 多文档查询多拿；
- 冗余控制：去重（片段重叠度 > 0.4 就合并 / 丢弃），避免重复内容占用 token；
- 评估指标：召回率、问答准确率，A/B 测试不同 Top (3/5/7)，看幻觉率、答案完整性。

面试口述精简版：
上下文条数不是固定值，由模型 token 预算决定。首先扣除系统提示词、用户 query 和回答预留 token，剩下的预算除以单切片平均长度，算出最大可容纳条数。检索出来先用向量粗召回，再 rerank 重排，优先保留高分片段，过滤重复、低相关内容；简单问题减少条数，复杂问题动态增加，同时做 A/B 实验，看幻觉和回答完整度来确定最终取值。

2. 切片长度（Chunk size）怎么选？重叠 overlap 怎么定

面试官想看：你知道 trade-off，不是直接说 chunk=512

核心权衡

chunk 太小：语义割裂，一段完整故障流程被拆碎，检索召回不完整，丢失上下文；向量颗粒细，但是语义碎片化
chunk 太大：单块包含很多无关信息，向量噪声大，容易引入无关内容，挤占 token，增加幻觉

选型原则

根据文档类型来定
- 故障工单、runbook、短运维文档：256~512 token
- 方案文档、架构说明、长故障复盘报告：512~1024 token
- 代码 / 日志：更小，128~256，按语句 / 函数块切，不要硬切句子中间
重叠 overlap：一般取 chunk size 的 10%~20%
作用：保证跨 chunk 的连续语义不会被切断（比如一条故障原因和解决方案被切在两个块）。overlap 太高会大量重复、浪费 token；太低会丢失跨块上下文。
两种切片方式对比
- 固定大小切片（简单，工程常用）：缺点会把一句话、一个完整逻辑块拦腰切断
- 语义切片（推荐生产）：优先在段落 / 章节边界切割，保证 chunk 内部语义完整，再限制最大 token 上限
验证方式：做召回测试，看关键信息是否被完整召回；看问答幻觉率变化。

口述精简：
切片长度没有万能值，核心是平衡语义完整性和向量噪声。我会按文档类型区分：运维 runbook、故障复盘一般 512token，重叠 10%-20%；日志、简短工单用 256。优先语义边界切，尽量不切断完整逻辑。chunk 太小语义断裂，召回不全；chunk 太大混入无关内容，增加幻觉。上线前做召回评估，对比不同 chunk 的召回指标。

3. 如何准确判断用户意图

区分：意图分类（用户想干嘛）≠ 语义检索（找相关文档）

思路，分 3 层判断

先定义意图字典（先枚举业务内所有意图，面向 CRE 场景）
CRE / 运维 Agent 常见意图：故障咨询、资源查询、SLO 指标查询、工单提交、变更咨询、权限申请、咨询产品架构、闲聊、无法识别
实现方式（3 种方案，从低成本到高成本）
① 小样本：Prompt 提示 + Few-shot 示例，直接让 LLM 分类意图（快速上线，适合意图少）
② 中等规模：微调小分类模型（text-classification），输出意图 + 置信度，置信度低于阈值交给 LLM 兜底
③ 复杂场景：多轮对话状态维护，结合对话历史判断意图（用户多轮追问，意图会变化）
准确性保障手段
- 设置置信度阈值：低于阈值 → 拒识，反问澄清，不强行猜测意图（非常重要，减少幻觉）
- 意图归一化：同义问句映射到同一个意图，比如 “服务延迟高”、“访问很慢” 归为【故障咨询】
- 区分主意图 + 子意图：主意图是故障排查，子意图是查延迟 / 查报错码
- 负样本：加入闲聊、无关问题样本，避免误识别
- 迭代：收集线上误判 case，补充 few-shot 样本，迭代优化

口述精简：
判断意图，第一步先梳理业务内全部意图清单。实现上可以用 Few-shot 大模型分类，或者微调轻量分类模型，输出意图和置信度。置信度低于阈值时不强行判定，主动向用户澄清。同时结合对话上下文，区分主意图和子意图；持续收集线上误识别 case，迭代 prompt 或者训练样本，提升准确率。

4. Agent 是如何识别意图的？（重点，区分 Agent 和普通 RAG 问答）

很多人踩坑：把 Agent 意图识别等同于简单文本分类。面试官想听到：Agent 是状态驱动 + 工具选择，不是一次性分类。

核心一句话：

普通 RAG：一次性识别意图，检索文档回答。
Agent 的意图识别是多轮、动态、带工具感知，分为 2 个阶段：

初始意图解析（第一轮用户输入）
模型理解用户目标，判断：用户是直接问答，还是需要调用工具（查询监控指标、查询工单、调用 API 查资源、执行检索）。
本质是：LLM 理解用户需求，输出规划 + 工具选择，而不是单纯标签分类。
动态意图追踪（多轮对话）
Agent 维护对话状态、已获取信息。用户后续提问，不是重新做一次独立意图分类；而是基于上一轮工具返回结果 + 历史对话，重新评估当前目标。
举例：
用户：集群延迟高 → Agent 识别意图：故障排查，调用监控 API 查指标
工具返回：region = 西南区域，接口 5xx 上涨
用户继续问：这个故障影响哪些客户
Agent不是重新做独立分类，基于上下文，识别意图：基于刚才的故障数据，查询受影响客户列表，选择客户查询工具。
底层机制拆解

依靠系统 Prompt：告诉 Agent 有哪些可用工具、每个工具能干什么、入参是什么；
LLM 做推理：分析用户问题，判断是否需要调用工具，调用哪个工具，提取参数；
工具返回结果后，Agent 再次思考：信息够不够回答用户？要不要继续调用其他工具，还是直接整理答案；
异常分支：需求模糊，缺少参数 → Agent 主动追问用户获取参数；需求超出工具能力 → 拒绝。

口述精简版（面试直接说）：
Agent 的意图识别和简单文本意图分类不一样，是动态多轮的。
首先系统 prompt 会告诉 Agent 可用工具列表，第一轮用户提问，大模型推理判断用户目标，选择对应的工具并提取参数；调用工具拿到返回数据后，结合历史对话，重新评估当前意图，判断信息是否充足，决定继续调用工具或者直接回答。如果信息缺失，Agent 会主动追问参数。整个过程会持续维护对话上下文状态，每一轮的意图判断都依赖上一轮的工具结果，而不是孤立地对单句做分类。

5. 如果意图识别置信度低，你怎么处理？

考点：避免强行猜测意图导致幻觉、误调用错误工具（生产事故风险）

回答思路

先设置置信阈值，低于阈值进入「拒识分支」，不强行打标签、不调用工具
区分两种低置信场景：
- 场景① 用户需求描述模糊，信息不足：主动澄清提问，提取缺失参数。
  例：用户说 “服务有问题”，没有指定集群 / 区域。Agent 反问：麻烦告诉我是哪个区域集群、什么报错现象？
- 场景② 问题不在业务支持意图范围内（闲聊、无关问题）：礼貌告知当前 Agent 能力边界，拒绝处理
兜底降级策略：
- 方案 A：交给人工接管（CRE 工单场景：自动生成工单转给运维 / 客户工程师）
- 方案 B： fallback 到通用 RAG 问答，禁止调用写操作类高危工具（比如变更、资源删除 API，高危工具必须严格禁止在低置信下触发）
数据闭环：所有低置信 case 入库，作为负样本 / 难例，定期迭代 prompt、few-shot 样本，或者补充到分类模型训练集，持续优化。

✅口述精简：
我会预先设定置信阈值，低于阈值就不强行判定意图，也不调用高危工具。先判断是用户描述模糊还是超出能力范围：信息缺失就主动追问获取参数；超出业务范围则告知能力边界。同时设置降级，低置信场景不能调用变更、删除这类高危 API，可以降级为纯知识库问答或者转人工。最后收集这类 case 持续优化意图识别样本。

6. chunk 切分如果刚好把关键故障原因切到下一个块，怎么优化？

考点：切片边界问题，RAG 经典痛点，面试官想看你不止知道 overlap，有多层优化方案

回答思路

基础：调大 overlap。比如原本 10%，提升到 15~20%，缓解跨 chunk 语义断裂；但 overlap 不能无限大，否则重复内容占用 token、引入冗余噪声。
优化切分策略：放弃固定 token 硬切，改用语义感知切片
按文档天然边界切割（标题、段落、章节、故障条目），优先保证完整逻辑单元在同一个 chunk，再限制最大 token 上限。运维文档、runbook 特别适合。
后处理增强：
- 文档预处理：把故障「现象 - 根因 - 解决方案」绑定成结构化块，提取摘要 + 关键字，存入元数据；检索时可以通过元数据关联，即使被切成两块，也能同时召回关联片段。
- 检索后合并：召回结果里面，检测片段之间来源文档 + 相邻位置，如果是连续 chunk，自动合并后再送入 LLM。
召回侧兜底：Rerank 重排模型，能识别跨分片的相关片段，提升相邻 chunk 排序优先级。
评估验证：构造测试集，专门放这种跨分片案例，做召回评测，观测关键信息召回率。

✅口述精简：
首先可以适度提高重叠窗口，但不能过大避免冗余。更核心是改用语义切片，在段落、章节边界切割，尽量不打断故障的现象、根因、解决方案这个完整逻辑单元。同时在文档预处理阶段增加元数据，把关联信息绑定；检索到相邻分片可以自动合并，配合 rerank 提升相邻相关片段分数。最后构造专项测试用例，验证跨分片场景的召回效果。

7. TopN 取太多，token 溢出怎么处理？

考点：token 预算管理，动态控制，不是简单截断

回答思路

前置动态 TopN（优先方案）：不是固定条数，基于剩余 token 预算动态计算最多能塞多少片段
计算：可用 token = 模型窗口 - 系统 prompt - 用户 query - 预留输出 token；用这个预算反推可容纳 chunk 数量，动态调整召回条数。
检索后裁剪策略：
- 先 rerank 排序，保留高分片段，优先丢弃低分、低相关片段；
- 做片段去重：相似度高的重复内容合并，减少 token 占用；
- 禁止粗暴对半截断单个 chunk：优先整体剔除低分完整 chunk，而不是把一个片段拦腰切断（截断容易产生残缺信息、幻觉）。
备选方案：摘要压缩。对高分长 chunk 做文档摘要，用摘要代替原文，节省 token；适合超长复盘报告。
极端兜底：预算不足，同时信息缺失。Agent 主动告知：参考资料较多，受上下文限制无法全部带入，询问用户是否缩小查询范围。

✅口述精简：
我不会写死 TopN，而是基于模型上下文预算动态计算最大可加载片段数量。检索结果经过 rerank 排序，优先剔除低分完整片段，同时做片段去重，尽量不截断单条 chunk。如果内容还是超，可以对长文档做摘要压缩减少 token 消耗。极端情况下，主动提示用户缩小查询范围，避免信息过载和幻觉。

8. 怎么评估意图识别好坏，用什么指标？

考点：区分分类指标，还要结合 Agent 业务指标（不只是 Accuracy）

回答思路

分为离线评估指标 + 线上业务指标

离线指标（标注测试集）
- Accuracy 准确率：整体判断正确比例
- Precision 精确率：判定为 A 意图的样本里，真正是 A 的比例（重点！运维 Agent，防止误判，误调用工具，精确率优先级高于召回）
- Recall 召回率：真实 A 意图，被识别为 A 的比例
- F1 综合值；混淆矩阵，重点看哪两类意图容易混淆（比如故障咨询 vs 指标查询）
- 置信度分布：统计低置信样本占比
线上业务指标（Agent 场景特有，面试官最爱听）
- 误调用率：识别意图错误，调用了错误工具（高危指标，CRE 运维场景核心）
- 澄清轮次：用户需要被 Agent 反问澄清的对话占比，越高代表意图识别越差
- 人工接管率：意图识别失败，流转人工处理的比例
- 用户满意度、对话完成率
评估流程：
构建标注测试集，包含正样本、同义问句、模糊问句、负样本（闲聊）；定期跑回归测试，迭代模型 /prompt 之后对比指标变化。持续采集线上 case 做抽样标注。

✅口述精简：
离线用标注数据集，统计准确率、精确率、召回率和 F1。运维 Agent 场景里精确率更关键，防止意图识别错误导致误调用工具。线上重点看业务指标：工具误调用率、需要澄清的对话占比、人工接管率。我们会持续收集线上 case 抽样标注，每次迭代 prompt 或模型都做回归测试，对比指标变化。

9. Rerank 和向量检索的区别，为什么不能只用向量检索？

向量检索是粗召回，基于向量空间相似度，擅长语义匹配，但容易出现语义假象（语义相近但内容无关）。Rerank 是精细排序，会理解文本细节，区分字面相似和真正相关，把真正有用的片段提升到前排。向量召回保证召回率，Rerank 提升精确率，二者配合。只用向量检索很容易引入大量看起来像、实际无关的文档，挤占 token，产生幻觉。

10. 幻觉怎么排查和治理（CRE 场景，客户故障回答幻觉很致命）

源头：检索质量（chunk、召回、rerank，减少无关资料进入上下文）
Prompt 约束：强制要求答案只能从提供的参考文档输出，不知道就说无法确认，禁止编造；标注引用来源
后校验：Agent 输出后做事实校验，核对关键参数、故障编号、SLO 数据是否和原文匹配
评估：构造幻觉测试集，统计幻觉率；线上收集客户反馈的幻觉 case，回流优化知识库

11. Agent 工具调用的参数提取失败怎么处理？

先看参数是否在对话历史内，尝试重新提取；
缺失必填参数：主动向用户追问；
参数格式错误：提示模型修正格式，重试 1~2 次；
多次提取失败：终止工具调用，降级问答或转人工，避免调用 API 报错甚至误操作。

12、为什么 LLM 选 DeepSeek V2，不选通义千问 (Qianwen)、OpenAI

口述主回答

我们金融质检 Agent 项目，有几个硬性约束：数据不能出境、Function Call 工具调用能力要强、长文本支持、私有化本地部署选项、可控成本。

OpenAI GPT 系列（GPT3.5/GPT4）不选用的原因
- 最大卡点：数据出境合规风险。项目里是金融录音 ASR 文本、客户工单、监管规则，属于敏感业务数据，不能传到海外 OpenAI 接口，这是一票否决项。
- 其次：API 调用成本高，大批量质检任务，token 费用会持续上涨；网络链路不稳定，有跨境延迟、超时风险，对 Agent 多轮 tool 调用链路不友好。
- 补充：虽然 GPT 的指令遵循、工具调用效果很好，但合规红线无法突破，直接排除。
对比通义千问 Qianwen
通义千问也有国内公有 API、私有化版本，中文能力不错，但当时评估有几点短板：
- 早期版本Function Call（工具调用）稳定性弱于 DeepSeek V2。我们 Agent 大量依赖多轮循环 tool 调用（查工单→算风险分→更新任务），LLM 如果经常输出非法 JSON、参数漏填，会直接造成 Agent 链路失败。DeepSeek V2 对 function call 做了专项优化，JSON 格式输出的准确率更高。
- 长上下文性价比：DeepSeek V2 支持更大上下文窗口，处理长 ASR 转写录音文本（几千字），长文本召回后的规则理解能力更好；同等上下文长度，推理成本更低。
- 私有化部署灵活性：DeepSeek 权重开源，我们可以本地私有化部署，完全隔绝业务敏感数据；通义千问私有化需要走阿里云商业授权，交付周期和约束更多。
- 劣势客观说（显得客观，不踩竞品）：通义千问通用对话的润色能力很强，适合普通知识库问答；但我们场景核心诉求不是文案润色，是稳定的结构化输出、工具调用、长文本规则推理，DeepSeek 更匹配。

✅ DeepSeek V2 选型优势（贴合你的 Agent 项目）

开源权重，支持私有化部署，敏感金融数据不出内网，满足合规；
Function Call 能力强，结构化 JSON 输出稳定，适配 Agent 多轮工具编排，减少解析失败；
长上下文能力强，适合处理 ASR 长文本 + 多条 RAG 分片，做监管条款推理；
推理成本可控，自托管之后，只消耗 GPU 算力，没有按 token 持续计费；
中文理解能力足够，适配中文监管文档、中文录音转写文本。

一句话精简口述版（面试快速说）
核心约束是金融敏感数据不能出境，直接排除 OpenAI。对比通义千问，DeepSeek V2 的 Function Call 结构化输出更稳定，长文本推理能力更适合我们的长 ASR 质检文本；并且支持私有化部署，数据不出内网，满足合规要求。我们场景重点不是日常对话，而是工具调用、结构化输出、规则推理，所以选 DeepSeek V2。

面试官追问：DeepSeek V2 有没有缺点？

回答：
有缺点。通用创意写作能力不如 GPT4，基础常识能力略弱于通义千问；同时自托管需要维护 GPU 集群，增加运维成本。所以选型是取舍：牺牲一部分通用对话能力，换取私有化、更强的工具调用、数据安全。

13、Embedding：为什么 BGE-Large-ZH，不用 OpenAI text-embedding

注意：是 BGE-Large-ZH（BAAI/bge-large-zh）

口述主回答

RAG 向量模型选型，核心评估维度：中文语义理解、向量空间一致性、部署成本、数据合规、检索效果。

OpenAI embedding 不选的核心原因
- 同样合规问题：知识库文档、监管规则、ASR 文本是金融敏感内容，不能上传到 OpenAI 接口做向量化，数据出境风险。
- 额外调用成本：大批量文档分片入库，百万级 chunk，调用 OpenAI Embedding 接口会产生大量 token 费用；文档更新、增量同步时反复调用，成本持续增加。
- 向量空间黑盒：OpenAI 向量模型闭源，无法复现向量结果；模型版本升级，向量空间会发生变化，旧向量全部失效，需要全量重刷 Milvus，风险很高。
BGE-Large-ZH 选型优势
- 专门针对中文语料训练。项目素材是中文监管文档、中文录音文本，中文段落、专业术语理解效果显著优于 OpenAI embedding（OpenAI embedding 原生是英文优先，中文语义召回会偏弱）。
- 开源本地部署，向量化在内网完成，原始文档不会流出，满足金融数据合规。
- 向量可控，权重固定，版本稳定，向量空间不会无感知变更；可以本地全量复现召回效果，方便调试 RAG 链路。
- 支持向量归一化输出，和 Milvus 向量检索（余弦相似度）完美适配。
- 工程友好：Sentence-Transformers 生态成熟，分片批量向量化很方便。

客观劣势（一定要讲，显得客观）
BGE-Large-ZH 推理速度比 API 慢，需要 GPU 做批量编码；长 chunk 的编码性能一般，所以我们会提前控制分片长度，做文本清洗。

一句话精简口述版：

OpenAI Embedding 闭源，存在敏感数据出境合规风险，大批量文档向量化成本高，向量空间不可控。BGE-Large-ZH 是中文专项训练的开源向量模型，中文知识库召回效果更好，可以内网私有化部署，向量可复现，适配我们中文监管文档 RAG 场景，所以选用。

14、配套代码片段（BGE embedding，面试可以提）

from sentence_transformers import SentenceTransformer
import numpy as np

# 加载BGE-large-zh
embedding_model = SentenceTransformer("BAAI/bge-large-zh")

def get_text_embedding(text: str) -> list[float]:
    # bge推荐开启归一化，余弦相似度检索
    vec = embedding_model.encode(
        text,
        normalize_embeddings=True,
        convert_to_numpy=True
    )
    return vec.tolist()

# 文档分片入库
chunk_text = "监管规则：禁止向客户承诺保本收益"
vec = get_text_embedding(chunk_text)
# milvus.insert(chunk_id, vector=vec, chunk_text=chunk_text, tenant_id="xxx")

# 用户query检索，**必须同一个模型**
query = "录音中业务员是否承诺保本？"
query_vec = get_text_embedding(query)
# milvus.search(query_vec, top_k=3)

15、高频衍生追问（面试官顺着深挖，全部准备好）

Q1：BGE 有很多版本，bge-base、bge-large，为什么选 large 而不是 base？

回答：
large 参数量更大，长中文段落、专业领域文本（金融监管条款）语义区分度更好，RAG 召回准确率更高。
代价：显存占用更高，推理更慢。我们知识库体量不是亿级，GPU 资源可以承受，优先保证召回精度；如果是高并发轻量问答场景，base 足够。

Q2：有没有评估其他中文 Embedding？比如 text2vec、通义千问 embedding？

回答：
评估过。text2vec 轻量、速度快，但复杂金融专业文本召回精度弱于 BGE。通义千问 embedding 公有 API 效果不错，但公有 API 同样存在数据上传风险；私有化版本需要商业授权。
我们场景优先中文语义召回精度 + 私有化可控，最终选定 BGE-large-zh。

Q3：DeepSeek V2，你是 API 调用还是本地私有化推理？GPU 资源怎么评估？

回答：
采用私有化部署，模型做量化（AWQ 量化 4bit）降低显存占用，部署在内部 GPU 服务器。
评估：单卡 A10 可以承载一定并发；Agent 是 IO 偏多业务，LLM 推理是瓶颈，做了请求队列 + 并发限流，防止 GPU 过载。

Q4：如果 DeepSeek 的 function call 偶尔解析失败，你怎么兜底？

回答：
两层兜底：

Prompt 增强：给 LLM 增加强指令，强制严格输出 JSON，禁止多余解释文字；
代码层：捕获 JSON 解析异常，二次重试，附带纠错 prompt；重试失败，直接降级，不再继续 tool 调用。

Q5：模型量化带来什么影响？4bit 量化会不会严重降低 Function Call 和推理精度？

回答：
4bit AWQ 量化，对结构化输出、中文规则推理影响很小，能够满足我们质检业务。
但会轻微损失长文本细节理解能力；我们提前做了业务验证，对比原模型，工具调用成功率下降幅度在可接受范围，显存节约非常明显，是算力和精度之间的取舍。

Q6：为什么不把 Embedding 和 LLM 放在同一个 GPU 上？

回答：
做了资源隔离。Embedding 是批量任务，高吞吐；LLM 推理是流式单次请求，低吞吐、延迟敏感。
混跑会抢占显存、GPU 计算资源，互相影响延迟。独立分开部署，各自扩缩容，方便压测和故障隔离。

Q7：模型版本升级，向量 / LLM 需要注意什么？

Embedding：模型版本变更，向量空间改变，旧 Milvus 向量全部失效，必须全量重新向量化入库。升级前要做灰度验证，对比召回指标。
LLM：升级前做一套业务评测集，验证 Function Call 成功率、幻觉率、规则推理准确率，指标达标才上线。

转载自 CSDN-专业IT技术社区

原文链接：https://blog.csdn.net/m0_55336994/article/details/167920716