极豆科技 · AI-Agent面经 · 一面 · 2026-03-13 · mj-nc-2026-1927

1、简单讲一下，MCP 是什么？它和 function call 有什么区别？

MCP，一般指 Model Context Protocol，本质上可以把它理解成一种让大模型和外部工具、外部数据源、外部能力进行标准化连接的协议。它不是单纯某一个函数调用接口，而是想解决“大模型怎么以统一方式发现工具、读取资源、调用能力、获取上下文”这个问题。

如果只说直白一点，MCP 更像是给大模型接外部世界时定义了一套“通用插口”。比如一个 Agent 想访问数据库、文件系统、知识库、内部服务，如果每接一个系统都单独写一套协议，维护成本会很高。MCP 的思路就是把这些能力抽象成统一的协议层，让模型或 Agent 框架可以按同样方式去发现和使用它们。

而 function call 更像什么？更像是 LLM 输出一个结构化调用意图，然后由宿主程序去执行某个预先注册好的函数。重点在“函数调用”本身。也就是说，function call 主要关注的是：模型决定调用哪个函数、传什么参数、再把执行结果喂回模型。

两者的区别可以这么讲：

MCP 更偏协议层、生态层、标准化连接层；function call 更偏模型调用工具时的一种具体机制。

所以 function call 更像一个“动作”，而 MCP 更像一套“规则和接口标准”。如果再说得更面试一点：

function call 解决的是“模型怎么调函数”
MCP 解决的是“模型和外部能力怎么标准化互通”

你可以把 function call 看成 MCP 体系里可能承载的一种能力形式，但 MCP 本身不等于 function call。

2、为什么会出现并行调用？保单查询、理赔查询不是应该先做意图判断吗？

正常情况下，确实很多系统会先做意图识别。比如先判断用户到底是在查保单、查理赔，还是查投保状态。如果意图非常明确，而且路由准确率高，那最省资源的方式当然是：先路由，再只调一个服务。

但实际项目里出现并行调用，往往有几个原因。

第一个原因是用户意图可能不干净。用户一句话里可能同时包含多个意图，比如：

“帮我看看这个保单现在状态怎么样，之前那次理赔有没有到账”

这时候就不是单一意图了，而是保单查询和理赔查询都相关。如果还坚持只走单路，很容易漏信息。

第二个原因是路由模型不是 100% 准。如果系统前面意图识别置信度不高，直接并发调用两个轻量查询服务，最后再聚合结果，有时候反而更稳。本质上是拿一点资源换召回和鲁棒性。

第三个原因是有些查询链路本身成本低、耗时短。比如查保单和查理赔底层都是读接口，而且响应时间差不多，那并发发出去再聚合结果，整体 RT 反而更短。

所以这题更成熟的回答不是“并行一定对”或者“先意图一定对”，而是：

是否并行，取决于意图识别的置信度、服务调用成本、用户问题是否可能多意图，以及整体延迟目标。

代码上，最常见就是 asyncio.gather：

import asyncio

async def query_policy(user_id):
    await asyncio.sleep(0.2)
    return {"type": "policy", "data": "保单状态：生效中"}

async def query_claim(user_id):
    await asyncio.sleep(0.3)
    return {"type": "claim", "data": "理赔状态：审核通过"}

async def main():
    user_id = "u1001"
    policy_result, claim_result = await asyncio.gather(
        query_policy(user_id),
        query_claim(user_id)
    )
    print(policy_result)
    print(claim_result)

asyncio.run(main())

3、多用户也可以用 async gather 一起调吗？

可以，但这里要分清楚两个层面。asyncio.gather 本身可以把多个协程并发执行，所以从语义上讲，它当然可以同时处理多个用户请求里的多个异步任务。但这不等于“所有多用户请求都丢给一个 gather 就完事”。

如果是单个用户请求内部有多个 I/O 操作，比如查保单、查理赔、查用户画像，这时候 gather 很适合。如果是多个用户同时访问系统，通常是 Web 框架本身负责并发接入，比如 FastAPI、aiohttp、uvicorn 这些异步服务框架。每个请求进来后，在请求内部再决定要不要 gather 多个下游任务。

也就是说：

多用户并发：主要靠服务框架、事件循环、协程调度
单请求内多任务并发：常用 asyncio.gather

有一个很重要的注意点是，gather 适合 I/O 密集型任务，比如查库、调接口、读 Redis、访问远程服务。如果你拿它去跑纯 CPU 密集任务，比如大规模本地推理、复杂计算，就不一定有效，因为 Python 解释器和事件循环不是这么用的。这种情况一般要考虑线程池、进程池，或者直接把重任务下沉到独立服务。

另外，多用户并发时还要注意几个现实问题：数据库连接池够不够、Redis 连接数够不够、下游接口会不会被打挂、单个请求里的并发数要不要限流。所以答案不能只停留在“可以”。

代码

import asyncio

async def handle_one_user(uid):
    async def query_a():
        await asyncio.sleep(0.1)
        return f"{uid}-A"

    async def query_b():
        await asyncio.sleep(0.2)
        return f"{uid}-B"

    a, b = await asyncio.gather(query_a(), query_b())
    return {"uid": uid, "result": [a, b]}

async def main():
    results = await asyncio.gather(
        handle_one_user("u1"),
        handle_one_user("u2"),
        handle_one_user("u3"),
    )
    print(results)

asyncio.run(main())

4、简单讲一下你这个项目的生命周期，就是一次用户请求一共会经过系统哪几部分？

如果是 AI-Agent 项目，一般可以按“接入层 -> 编排层 -> 工具层 -> 数据层 -> 结果生成层 -> 返回层”去讲。

比较自然的讲法可以是这样：

用户请求先进 API 网关或者 Web 服务入口，先做鉴权、参数校验、请求日志和 trace_id 注入。然后进入 Agent 主流程，先做会话态读取、上下文拼接、用户信息加载。接着做意图识别或者任务路由，判断这次是普通闲聊、知识库问答、保单查询、理赔查询，还是多工具协同任务。

如果命中工具调用场景，就进入工具编排层。这一层会决定调用哪些内部服务，比如保单服务、理赔服务、Redis 缓存、MySQL、向量检索、规则引擎等。如果可以并发，就并发查；如果有依赖关系，就串行执行。

工具结果回来之后，进入结果聚合和后处理层。这里会做数据清洗、字段统一、缺失值处理、格式规整，必要时还会做 prompt 组装，把结构化结果和上下文拼成给大模型的输入。然后由 LLM 负责生成最终面向用户的自然语言回复，或者直接返回结构化结果。

最后在响应阶段，还会做日志记录、指标上报、缓存写回、会话落库、审计记录这些动作。如果系统做得更完整，还会有超时降级、兜底话术、敏感信息脱敏和失败重试。

5、连接池主要起什么作用，为什么要用它？

连接池最核心的作用，是复用连接，减少频繁创建和销毁连接的成本。不管是 MySQL、Redis，还是 HTTP 长连接，本质上建立连接都不是零成本的。如果每来一个请求就新建连接，用完马上关闭，吞吐量一上去，系统开销会非常大。

连接池就是预先维护一批可复用连接。请求来了，从池子里取一个连接；用完后不是销毁，而是归还池子，后面继续复用。这样做最大的好处有三个。

第一是性能更好。少了反复建立 TCP 连接、认证握手这些成本，请求延迟会更稳。

第二是资源更可控。连接数不是无限涨，而是有上限。这样可以避免高并发时把数据库直接打爆。

第三是更方便做统一管理。比如连接超时、空闲回收、健康检查、最大连接数、等待队列，这些都可以放到连接池层统一处理。

所以你可以把连接池理解成：它既是性能优化手段，也是资源保护机制。

如果只说一句“提高性能”，面试官一般会继续追问。你最好顺手补充：

连接池不只是为了快，也是为了控连接数、减少系统抖动、提高高并发场景下的稳定性。

示例代码，以 aiomysql 为例：

import asyncio
import aiomysql

async def main():
    pool = await aiomysql.create_pool(
        host='127.0.0.1',
        port=3306,
        user='root',
        password='123456',
        db='test',
        minsize=1,
        maxsize=10
    )

    async with pool.acquire() as conn:
        async with conn.cursor() as cur:
            await cur.execute("SELECT 1")
            result = await cur.fetchone()
            print(result)

    pool.close()
    await pool.wait_closed()

asyncio.run(main())

6、MySQL 的事务可以讲一下吗？

事务就是把一组数据库操作当成一个整体来执行，要么都成功，要么都失败回滚，不允许只做一半。它主要是为了解决多个 SQL 操作之间的一致性问题。

经典特性就是 ACID：

A 是 Atomicity，原子性。事务里的多条操作不可再分，要么全部提交，要么全部回滚。

C 是 Consistency，一致性。事务执行前后，数据库要从一个一致状态变成另一个一致状态，不能破坏业务约束。

I 是 Isolation，隔离性。多个事务并发执行时，彼此之间要尽量互不干扰。

D 是 Durability，持久性。事务一旦提交，结果就要持久保存下来，即使数据库宕机，恢复后数据也应该还在。

举个很经典的例子，转账：

A 账户减 100
B 账户加 100

这两个操作必须放在同一个事务里。如果只扣了 A，B 没加成功，那数据就错了。所以事务存在的意义，就是保证这种场景下的数据正确性。

SQL 示例：

START TRANSACTION;

UPDATE account SET balance = balance - 100 WHERE user_id = 1;
UPDATE account SET balance = balance + 100 WHERE user_id = 2;

COMMIT;

如果中途失败，就：

ROLLBACK;

7、事务隔离级别你了解吗？

事务隔离级别，解决的是并发事务之间“互相看见到什么程度”的问题。MySQL 里常见的四种隔离级别是：

Read Uncommitted，读未提交

一个事务可以读到另一个事务还没提交的数据。这会带来脏读问题。也就是别人还没正式提交的数据，你先看到了，结果对方一回滚，你刚才读到的内容其实是无效的。

Read Committed，读已提交

只能读到其他事务已经提交的数据。这样可以避免脏读。但同一个事务里两次读取同一条数据，结果可能不一样，这叫不可重复读。

Repeatable Read，可重复读

同一个事务里多次读取同一条记录，结果保持一致。MySQL InnoDB 默认就是这个级别。它可以避免脏读和不可重复读。但理论上还可能有幻读问题，不过 InnoDB 通过 MVCC 和间隙锁，在很多场景下把幻读问题也压得比较好了。

Serializable，可串行化

最高隔离级别。事务之间几乎按串行顺序执行，并发能力最差，但隔离最强。一般线上高并发系统很少随便开到这个级别，因为性能代价比较大。

面试里常见要你说并发问题：

脏读：读到别人未提交的数据
不可重复读：同一事务前后两次读同一行不一致
幻读：同一事务前后两次按条件查，记录条数变了

实际项目里一般会用 MySQL InnoDB 默认的 Repeatable Read，再结合业务层幂等、唯一索引、悲观锁或乐观锁，去处理并发更新问题，而不是只依赖数据库隔离级别兜底。

8、Redis 的 ZSET 用过吗？

用过。ZSET 就是 有序集合，它和普通 SET 最大的区别是：ZSET 里的每个 member 都会关联一个 score，Redis 会按 score 自动排序。所以它特别适合“既要去重，又要排序”的场景。

底层上你可以理解成：它既有集合“成员唯一”的特性，又有排序能力。常见使用场景非常多，比如排行榜、延迟队列、优先级任务、按时间排序的消息流、TopN 统计。

比如做排行榜时，用户 id 作为 member，分数作为 score：

ZADD game_rank 1001 userA
ZADD game_rank 880 userB
ZADD game_rank 1200 userC

查前 3 名：

ZREVRANGE game_rank 0 2 WITHSCORES

如果做延迟任务，也经常用 ZSET。把任务执行时间戳当 score，把任务 id 当 member，定时扫当前时间之前的任务取出来执行。

redis-py：

import redis
import time

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

# 排行榜
r.zadd("rank", {"userA": 100, "userB": 80, "userC": 120})
top_users = r.zrevrange("rank", 0, 2, withscores=True)
print(top_users)

# 延迟任务
now = int(time.time())
r.zadd("delay_queue", {"task_1": now + 10, "task_2": now + 20})

tasks = r.zrangebyscore("delay_queue", 0, int(time.time()), withscores=True)
print(tasks)

AI-Agent一面面经