阿里国际 AI应用开发实习二面

1. 自我介绍

2. 如何在工程上实现兜底机制？具体到代码层面怎么做？

答案：兜底机制不能只写一句“失败后重试”，要按失败类型分层。模型层可能超时、限流、返回格式错误；检索层可能无结果、召回低相关；工具层可能参数错误、权限失败、下游服务不可用；生成层可能答案无引用或不符合 schema。不同错误要有不同兜底，不能统一重跑。

工程上我一般会做三层兜底：第一层是重试，比如超时、临时网络错误可以指数退避重试；第二层是降级，比如大模型失败就切小模型，rerank 失败就用召回分数排序；第三层是安全返回，比如证据不足就拒答，工具失败就返回部分结果和失败原因。对于有副作用的工具，不能盲目重试，必须加幂等 key。

import time
from enum import Enum

class ErrorType(str, Enum):
    TIMEOUT = "timeout"
    RATE_LIMIT = "rate_limit"
    JSON_ERROR = "json_error"
    PERMISSION_DENIED = "permission_denied"
    NO_EVIDENCE = "no_evidence"

def retryable(error_type):
    return error_type in {ErrorType.TIMEOUT, ErrorType.RATE_LIMIT, ErrorType.JSON_ERROR}

def call_with_fallback(primary_fn, fallback_fn, max_retry=2):
    last_error = None

    for i in range(max_retry + 1):
        try:
            return primary_fn()
        except TimeoutError as e:
            last_error = e
            time.sleep(0.5 * (2 ** i))
        except ValueError as e:
            last_error = e
            break

    try:
        return fallback_fn()
    except Exception:
        return {
            "status": "failed",
            "message": "当前服务不可用，已触发兜底，但仍未成功",
            "error": str(last_error)
        }

3. 上下文压缩一般怎么做？怎么避免压缩后丢失关键约束？

答案：上下文压缩不是简单让大模型“总结一下历史对话”。如果直接摘要，很容易把约束、否定信息、用户偏好、工具结果里的关键字段压没。比较稳的方式是结构化压缩，把历史内容分成任务目标、用户约束、已确认事实、工具结果、待办事项、被否定方案和风险点。

比如科研方案生成里，用户前面说“不要动物实验，只做细胞实验”，这个否定约束必须保留；用户说“预算不能超过 2 万”，这个硬约束也不能被摘要掉。压缩时我会要求模型输出固定 JSON，同时保留最近几轮原文。最终上下文一般是“最近窗口 + 结构化摘要 + 可检索长期记忆”。

def build_compressed_memory(history):
    prompt = f"""
    请把以下多轮对话压缩成结构化状态。
    必须保留：
    1. 用户目标
    2. 硬性约束
    3. 已确认事实
    4. 已否定方案
    5. 待办事项
    6. 工具调用结论

    对话历史：
    {history}

    输出 JSON。
    """
    return prompt

compressed_state = {
    "goal": "生成细胞实验方案验证某靶点作用",
    "constraints": ["不做动物实验", "预算不超过2万元", "周期不超过4周"],
    "confirmed_facts": ["用户关注炎症通路", "已有qPCR设备"],
    "rejected_options": ["动物模型", "高通量测序"],
    "pending": ["确认细胞系", "确认药物浓度梯度"]
}

4. 长上下文和压缩上下文分别适合什么场景？

答案：长上下文适合需要保留原文细节的任务，比如论文逐段分析、合同全文审查、代码仓库定位、长表格问答。它的优点是信息完整，缺点是成本高、延迟大，而且模型不一定真的能关注到中间的关键信息。

压缩上下文适合连续多轮任务和状态驱动任务，比如用户反复修改实验方案、持续讨论一个课题、Agent 多步执行。压缩后成本更低、状态更清晰，但风险是细节丢失。所以真实系统不会二选一，而是最近对话保留原文，远期历史压成结构化状态，必要时再从长期记忆里召回原文片段。

context_pack = {
    "recent_turns": [
        "用户：不要动物实验",
        "助手：可以改为细胞实验和分子实验"
    ],
    "compressed_state": {
        "hard_constraints": ["不做动物实验"],
        "current_plan": "细胞实验验证靶点作用"
    },
    "retrieved_long_term_memory": [
        "用户之前偏好低成本实验方案"
    ]
}

5. 其他上下文机制你了解哪些？

答案：除了滑动窗口和摘要压缩，还有几种常用机制。第一是结构化状态，把任务状态从聊天记录里抽出来，作为 Agent 的真实执行状态。第二是检索式记忆，把历史对话、用户偏好、工具结果向量化，按需召回。第三是分层记忆，把短期会话、任务级记忆、用户级长期记忆分开存。第四是事件溯源，把每次工具调用和状态变化保存成 event，方便回放和恢复。第五是上下文路由，不同任务只拿相关上下文，避免所有信息都塞给模型。

比较高级一点的是上下文预算分配。比如一次 prompt 最多 16k token，其中系统指令 1k，当前问题 1k，最近历史 3k，检索证据 8k，结构化状态 2k，剩下留给工具结果。这样不会因为历史太长挤掉证据。

budget = {
    "system_prompt": 1000,
    "current_query": 1000,
    "recent_history": 3000,
    "retrieved_evidence": 8000,
    "structured_memory": 2000,
    "tool_observation": 1000
}

6. 记忆系统怎么设计？短期记忆和长期记忆怎么划分？

答案：短期记忆解决当前会话连续性，长期记忆解决跨会话个性化和历史经验复用。短期记忆通常包括最近几轮对话、当前任务状态、工具调用结果和待确认事项，适合放 Redis 或数据库热表。长期记忆包括用户偏好、历史任务结论、常用实体、已确认背景信息，适合放数据库和向量库。

长期记忆不能什么都存，否则会污染后续回答。要有写入策略，比如只有用户明确确认的信息、反复出现的偏好、任务最终结论才写入长期记忆。长期记忆召回后也不能直接当事实，需要带来源、时间和置信度。

memory_item = {
    "user_id": "u_001",
    "memory_type": "preference",
    "content": "用户偏好低成本、短周期实验方案",
    "source": "conversation",
    "confidence": 0.86,
    "created_at": "2026-05-11",
    "expire_at": None
}

7. 怎么让用户感受不到压缩记忆带来的时间开销？

答案：压缩记忆不要阻塞主链路。用户每发一轮消息，如果都同步压缩完整历史，延迟会很明显。更好的方式是异步压缩加增量更新：当前回复先使用最近窗口和已有摘要，回复完成后后台异步把新一轮对话合并进摘要。下一轮请求再使用最新摘要。

工程上可以用消息队列或后台任务处理压缩。对于特别关键的状态，比如用户刚刚修改了硬约束，可以同步更新结构化状态；普通闲聊和低价值内容可以异步处理。这样既保证关键约束及时生效，又不让用户等待摘要生成。

from fastapi import BackgroundTasks

def update_memory_async(session_id, new_turn):
    # 后台调用模型，把 new_turn 合并进 compressed_state
    print("compress memory:", session_id, new_turn)

def chat(session_id, user_input, background_tasks: BackgroundTasks):
    context = load_context(session_id)
    answer = generate_answer(context, user_input)

    background_tasks.add_task(
        update_memory_async,
        session_id,
        {"user": user_input, "assistant": answer}
    )

    return answer

8. Redis 是什么？在大模型应用里一般用来做什么？

答案：Redis 是内存型 key-value 数据库，支持字符串、哈希、列表、集合、有序集合、Stream、Bitmap、HyperLogLog 等结构。它的特点是读写快、支持过期时间、原子操作和发布订阅，所以常被用作缓存、分布式锁、计数器、消息队列、限流器和会话存储。

在大模型应用里，Redis 很适合放短期会话状态、任务运行状态、流式输出缓冲、工具调用幂等 key、热点知识缓存、限流计数和异步任务队列。但 Redis 不适合做最终可信存储，重要数据还是要落 MySQL、PostgreSQL 或对象存储。

import redis
import json

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

session_state = {
    "session_id": "s_001",
    "current_goal": "生成实验方案",
    "stage": "collect_constraints"
}

r.setex(
    "session:s_001:state",
    3600,
    json.dumps(session_state, ensure_ascii=False)
)

print(json.loads(r.get("session:s_001:state")))

9. Redis 的数据类型以及适用场景？

答案：String 适合缓存单个值，比如 token、计数器、JSON 状态。Hash 适合存对象字段，比如会话状态、用户画像。List 适合简单队列和最近消息列表。Set 适合去重，比如已处理任务 ID。ZSet 适合排行榜、延迟队列、按时间排序的任务。Stream 适合更可靠的消息流和消费组。Bitmap 适合签到、布尔状态统计。HyperLogLog 适合 UV 估算。

在 Agent 系统里，我一般用 String 存压缩后的 session state，用 List 存最近 N 轮对话，用 ZSet 做延迟重试任务，用 Stream 做工具调用事件流，用 Set 存幂等 key，防止重复执行。

# 最近对话窗口
r.lpush("session:s_001:turns", json.dumps({"role": "user", "content": "不要动物实验"}))
r.ltrim("session:s_001:turns", 0, 9)

# 幂等 key
idempotency_key = "tool:submit_plan:req_123"
if r.set(idempotency_key, "1", nx=True, ex=600):
    print("第一次执行")
else:
    print("重复请求，直接拒绝或返回旧结果")