前端与 AI 应用开发高频手撕题：30 题可背诵答案

// 手撕版：Map 负责 O(1) 定位，双向链表负责 O(1) 调整访问顺序。
class Node<K, V> {
  prev: Node<K, V> | null = null;
  next: Node<K, V> | null = null;

  constructor(
    public key: K,
    public value: V,
  ) {}
}

class LRUCache<K, V> {
  private readonly cache = new Map<K, Node<K, V>>();
  // 哨兵节点避免在头尾插入、删除时写大量空指针判断。
  private readonly head = new Node<K, V>(undefined as K, undefined as V);
  private readonly tail = new Node<K, V>(undefined as K, undefined as V);

  constructor(private readonly capacity: number) {
    if (!Number.isInteger(capacity) || capacity <= 0) {
      throw new Error("capacity must be a positive integer");
    }
    this.head.next = this.tail;
    this.tail.prev = this.head;
  }

  get(key: K): V | undefined {
    const node = this.cache.get(key);
    if (!node) return undefined;
    this.moveToFront(node); // 命中后成为最新使用。
    return node.value;
  }

  put(key: K, value: V): void {
    const existing = this.cache.get(key);
    if (existing) {
      existing.value = value;
      this.moveToFront(existing);
      return;
    }

    const node = new Node(key, value);
    this.cache.set(key, node);
    this.addToFront(node);

    if (this.cache.size > this.capacity) {
      const leastRecent = this.removeLeastRecent();
      this.cache.delete(leastRecent.key);
    }
  }

  private addToFront(node: Node<K, V>): void {
    node.next = this.head.next;
    node.prev = this.head;
    this.head.next!.prev = node;
    this.head.next = node;
  }

  private remove(node: Node<K, V>): void {
    node.prev!.next = node.next;
    node.next!.prev = node.prev;
  }

  private moveToFront(node: Node<K, V>): void {
    this.remove(node);
    this.addToFront(node);
  }

  private removeLeastRecent(): Node<K, V> {
    const node = this.tail.prev!;
    this.remove(node);
    return node;
  }
}

# 手撕版：dict 负责 O(1) 定位，双向链表负责 O(1) 调整访问顺序。
class Node:
    def __init__(self, key=None, value=None):
        self.key = key
        self.value = value
        self.prev = None
        self.next = None


class LRUCache:
    def __init__(self, capacity: int):
        if not isinstance(capacity, int) or capacity <= 0:
            raise ValueError("capacity must be a positive integer")

        self.capacity = capacity
        self.cache = {}
        # 哨兵节点让头尾插入和删除不需要判空。
        self.head = Node()
        self.tail = Node()
        self.head.next = self.tail
        self.tail.prev = self.head

    def get(self, key):
        node = self.cache.get(key)
        if node is None:
            return None

        self._move_to_front(node)  # 命中后成为最新使用。
        return node.value

    def put(self, key, value):
        node = self.cache.get(key)
        if node is not None:
            node.value = value
            self._move_to_front(node)
            return

        node = Node(key, value)
        self.cache[key] = node
        self._add_to_front(node)

        if len(self.cache) > self.capacity:
            least_recent = self._remove_least_recent()
            del self.cache[least_recent.key]

    def _add_to_front(self, node):
        node.prev = self.head
        node.next = self.head.next
        self.head.next.prev = node
        self.head.next = node

    def _remove(self, node):
        node.prev.next = node.next
        node.next.prev = node.prev

    def _move_to_front(self, node):
        self._remove(node)
        self._add_to_front(node)

    def _remove_least_recent(self):
        node = self.tail.prev
        self._remove(node)
        return node

// 下一块从 size - overlap 处开始，保留上下文。
export function chunkText(text: string, size = 500, overlap = 80) {
  if (size <= overlap) throw new Error("size must exceed overlap");
  const result: string[] = [];
  for (let start = 0; start < text.length; start += size - overlap) {
    result.push(text.slice(start, start + size));
  }
  return result.filter(Boolean);
}

# step 必须为正数，否则切块不会前进。
def chunk_text(text: str, size: int = 500, overlap: int = 80) -> list[str]:
    if size <= overlap: raise ValueError("size must exceed overlap")
    step = size - overlap
    return [text[i:i + size] for i in range(0, len(text), step) if text[i:i + size]]

// 点积除以两个向量的模长，结果范围通常为 [-1, 1]。
export function cosine(a: number[], b: number[]) {
  if (a.length !== b.length || !a.length) throw new Error("invalid vectors");
  let dot = 0, aa = 0, bb = 0;
  for (let i = 0; i < a.length; i++) { dot += a[i] * b[i]; aa += a[i] ** 2; bb += b[i] ** 2; }
  return aa && bb ? dot / Math.sqrt(aa * bb) : 0;
}

import math
# zip 同步遍历两个同维向量。
def cosine(a: list[float], b: list[float]) -> float:
    if len(a) != len(b) or not a: raise ValueError("invalid vectors")
    dot = sum(x * y for x, y in zip(a, b))
    norm = math.sqrt(sum(x*x for x in a) * sum(y*y for y in b))
    return dot / norm if norm else 0.0

type Item = { id: string; vector: number[]; text: string };
// 先打分再排序；大规模数据应交给向量索引。
export function topK(query: number[], items: Item[], k = 3) {
  return items.map(item => ({ item, score: cosine(query, item.vector) }))
    .sort((a, b) => b.score - a.score).slice(0, k);
}

# 返回分数最高的 k 条，而不是只返回原始文档。
def top_k(query, items, k=3):
    ranked = [{"item": item, "score": cosine(query, item["vector"])} for item in items]
    return sorted(ranked, key=lambda row: row["score"], reverse=True)[:k]

// embed 与 llm 由调用方注入，避免绑定具体模型 SDK。
export async function answerWithRag(question: string, docs: Item[], embed: (s: string) => Promise<number[]>, llm: (p: string) => Promise<string>) {
  const hits = topK(await embed(question), docs, 3);
  const context = hits.map(({ item }) => item.text).join("\n---\n");
  return llm("只根据资料回答；资料不足时说明不知道。\n资料：\n" + context + "\n问题：" + question);
}

# 先检索再组织上下文，限制模型只基于资料回答。
async def answer_with_rag(question, docs, embed, llm):
    hits = top_k(await embed(question), docs, 3)
    context = "\n---\n".join(row["item"]["text"] for row in hits)
    prompt = f"只根据资料回答；资料不足时说明不知道。\n资料：\n{context}\n问题：{question}"
    return await llm(prompt)

type Tool = { description: string; run: (input: unknown) => Promise<unknown> };
// 白名单注册，阻止模型调用未授权函数。
const tools: Record<string, Tool> = {};
export function register(name: string, tool: Tool) { tools[name] = tool; }
export async function callTool(name: string, input: unknown) {
  if (!tools[name]) throw new Error("tool not allowed");
  return tools[name].run(input);
}

TOOLS = {}
# 仅暴露显式登记过的工具。
def register(name, description, fn): TOOLS[name] = {"description": description, "run": fn}
async def call_tool(name, input):
    if name not in TOOLS: raise ValueError("tool not allowed")
    return await TOOLS[name]["run"](input)

// 最大轮数避免模型反复调用工具造成死循环。
export async function runAgent(message: string, decide: (m: string) => Promise<{ type: "final"; text: string } | { type: "tool"; name: string; input: unknown }>) {
  let state = message;
  for (let turn = 0; turn < 6; turn++) { const next = await decide(state); if (next.type === "final") return next.text; state += "\n工具结果：" + JSON.stringify(await callTool(next.name, next.input)); }
  throw new Error("agent exceeded max turns");
}

# 将工具结果写回状态，供下一轮决策参考。
async def run_agent(message, decide):
    state = message
    for _ in range(6):
        next_step = await decide(state)
        if next_step["type"] == "final": return next_step["text"]
        result = await call_tool(next_step["name"], next_step["input"])
        state += "\n工具结果：" + repr(result)
    raise RuntimeError("agent exceeded max turns")

// 多个 worker 共享 cursor，因此同时最多执行 limit 个任务。
export async function withLimit<T>(jobs: Array<() => Promise<T>>, limit = 3) {
  const result: T[] = []; let cursor = 0;
  async function worker() { while (cursor < jobs.length) { const index = cursor++; result[index] = await jobs[index](); } }
  await Promise.all(Array.from({ length: Math.min(limit, jobs.length) }, worker)); return result;
}

import asyncio
# Semaphore 在 await 期间限制正在执行的任务数。
async def with_limit(jobs, limit=3):
    semaphore = asyncio.Semaphore(limit)
    async def run(job):
        async with semaphore: return await job()
    return await asyncio.gather(*(run(job) for job in jobs))

// 等待时间随失败次数翻倍；生产中还应加入随机抖动。
export async function retry<T>(fn: () => Promise<T>, retries = 3, baseMs = 300): Promise<T> {
  let error: unknown;
  for (let i = 0; i <= retries; i++) try { return await fn(); } catch (e) { error = e; if (i < retries) await new Promise(r => setTimeout(r, baseMs * 2 ** i)); }
  throw error;
}

import asyncio
# 只有临时错误才应该使用该重试器。
async def retry(fn, retries=3, base_ms=300):
    for i in range(retries + 1):
        try: return await fn()
        except Exception:
            if i == retries: raise
            await asyncio.sleep(base_ms * 2**i / 1000)

type Message = { role: string; content: string };
// 永远先保留 system，再从最新消息向前填充预算。
export function trimContext(messages: Message[], maxChars = 8000) {
  const system = messages.filter(m => m.role === "system");
  const rest = messages.filter(m => m.role !== "system"); let used = system.reduce((n, m) => n + m.content.length, 0); const keep: Message[] = [];
  for (let i = rest.length - 1; i >= 0; i--) { if (used + rest[i].content.length > maxChars) break; keep.unshift(rest[i]); used += rest[i].content.length; }
  return [...system, ...keep];
}

# 字符数只是最简估算；实际应使用模型 tokenizer。
def trim_context(messages, max_chars=8000):
    system = [m for m in messages if m["role"] == "system"]
    used, keep = sum(len(m["content"]) for m in system), []
    for message in reversed([m for m in messages if m["role"] != "system"]):
        if used + len(message["content"]) > max_chars: break
        keep.insert(0, message); used += len(message["content"])
    return system + keep

// 网络 chunk 可能截断事件，buffer 必须跨读取保留。
export async function* readSse(response: Response) {
  const reader = response.body!.pipeThrough(new TextDecoderStream()).getReader(); let buffer = "";
  while (true) { const { value, done } = await reader.read(); if (done) break; buffer += value; const events = buffer.split("\n\n"); buffer = events.pop()!;
    for (const event of events) { const data = event.split("\n").find(line => line.startsWith("data:"))?.slice(5).trim(); if (data && data !== "[DONE]") yield JSON.parse(data); }
  }
}

import json

# 只有遇到空行才得到一个完整 SSE 事件。
async def read_sse(response):
    buffer = ""
    async for chunk in response.aiter_text():
        buffer += chunk
        *events, buffer = buffer.split("\n\n")
        for event in events:
            line = next(
                (value for value in event.splitlines()
                 if value.startswith("data:")),
                "",
            )
            data = line[5:].strip()
            if data and data != "[DONE]":
                yield json.loads(data)