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

1. 自我介绍

2. RAG 的检索索引怎么设计？

答案：RAG 的检索索引不能只建一个向量字段就结束。实际系统里，我会把索引分成语义索引、关键词索引、结构化索引和权限索引。语义索引用来处理自然语言问题，关键词索引用来处理规则编号、商品编码、报关术语、单据字段名这类精确匹配，结构化索引用来按国家、业务类型、规则版本、生效时间过滤，权限索引用来保证用户只能查自己有权限的数据。

文档入库时也要保留足够元数据，比如 doc_id、chunk_id、parent_id、section_path、doc_type、country、effective_date、version、tenant_id。如果没有这些字段，后面很难做版本控制、权限隔离和证据回溯。RAG 的索引设计，本质上是为召回、过滤、重排和引用服务的。

chunk_doc = {
    "doc_id": "customs_rule_2026_001",
    "chunk_id": "customs_rule_2026_001_c12",
    "parent_id": "customs_rule_2026_001_sec3",
    "section_path": ["出口申报规则", "商品归类", "申报要素"],
    "doc_type": "policy",
    "country": "US",
    "version": "v2026.04",
    "effective_date": "2026-04-01",
    "tenant_id": "tenant_001",
    "content": "申报商品为电子元器件时，应填写品牌、型号、用途、材质等要素。",
    "embedding": [0.01, 0.03]
}

3. 怎么把一个普通工具变成 MCP Server？

答案：把普通工具变成 MCP Server，本质是把本地函数、数据库查询、HTTP 服务或者业务能力封装成标准化工具接口。关键不是简单暴露一个函数，而是要定义工具名、描述、输入 schema、输出 schema、错误格式、权限要求和超时策略。Agent 通过 MCP Client 发现工具，再根据工具描述和参数 schema 生成调用。

比如把“查询商品申报规则”变成 MCP 工具，需要定义输入参数：商品编码、目的国、业务类型；输出参数：规则列表、版本、生效时间、引用来源。工具内部仍然可以查数据库或搜索引擎，但对 Agent 来说，它看到的是一个稳定的协议化能力。

from mcp.server.fastmcp import FastMCP
from pydantic import BaseModel

mcp = FastMCP("customs-compliance-server")

class RuleQuery(BaseModel):
    hs_code: str
    country: str
    biz_type: str

@mcp.tool()
def query_declaration_rule(hs_code: str, country: str, biz_type: str) -> dict:
    """
    查询指定商品编码在目标国家下的申报规则。
    """
    # 实际项目中这里会查询数据库或检索服务
    return {
        "hs_code": hs_code,
        "country": country,
        "biz_type": biz_type,
        "rules": [
            {
                "rule_id": "R-2026-041",
                "content": "需填写品牌、型号、用途、材质、是否带电。",
                "version": "v2026.04"
            }
        ]
    }

if __name__ == "__main__":
    mcp.run()

4. LangGraph 的状态流转怎么设计？状态一般是什么类型？

答案：LangGraph 里状态通常会设计成一个结构化对象，可以用 TypedDict、Pydantic BaseModel 或普通 dict。状态里不应该只存消息列表，而应该存任务目标、当前步骤、用户输入、检索结果、工具调用结果、中间判断、错误信息和最终输出。这样图里的每个节点都可以读写状态，实现可恢复的流程编排。

比如审单场景中，状态可以包括：用户问题、单据字段、风险候选、检索证据、工具调用记录、重试次数和最终审查意见。节点之间通过状态传递，不是靠自然语言上下文猜测。这样比普通 chain 更适合复杂任务，因为它可以根据状态条件分支、循环、重试和中断。

from typing import TypedDict, List, Dict, Any
from langgraph.graph import StateGraph, END

class ReviewState(TypedDict):
    query: str
    document_fields: Dict[str, Any]
    evidence: List[dict]
    risk_items: List[dict]
    retry_count: int
    final_answer: str

def parse_document(state: ReviewState):
    state["document_fields"] = {
        "hs_code": "854239",
        "country": "US",
        "description": "integrated circuit module"
    }
    return state

def retrieve_rules(state: ReviewState):
    state["evidence"] = [
        {"rule_id": "R001", "content": "集成电路模块需补充品牌、型号和用途。"}
    ]
    return state

def generate_answer(state: ReviewState):
    state["final_answer"] = "该单据存在申报要素不完整风险，需补充品牌、型号和用途。"
    return state

graph = StateGraph(ReviewState)
graph.add_node("parse_document", parse_document)
graph.add_node("retrieve_rules", retrieve_rules)
graph.add_node("generate_answer", generate_answer)

graph.set_entry_point("parse_document")
graph.add_edge("parse_document", "retrieve_rules")
graph.add_edge("retrieve_rules", "generate_answer")
graph.add_edge("generate_answer", END)

app = graph.compile()

5. SSE 和 WebSocket 的区别是什么？大模型流式输出选哪个？

答案：SSE 是服务端单向推送，基于 HTTP，客户端建立连接后，服务端可以不断推送事件。它实现简单、兼容性好、适合大模型流式输出、日志推送、任务进度通知这类服务端到客户端的单向数据流。WebSocket 是全双工通信，客户端和服务端都可以主动发送消息，适合在线协作、实时聊天、多人编辑、实时控制这类高频双向交互。

如果只是大模型逐 token 输出，SSE 通常更简单稳定。因为用户发起一次请求后，服务端持续返回 token，不需要复杂的双向协议。如果是 Agent 运行过程中用户随时打断、动态补充参数、实时控制工具执行，WebSocket 会更灵活。

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import time

app = FastAPI()

def token_stream():
    tokens = ["正在", "检索", "规则", "并", "生成", "答案"]
    for t in tokens:
        yield f"data: {t}\n\n"
        time.sleep(0.3)

@app.get("/stream")
def stream():
    return StreamingResponse(token_stream(), media_type="text/event-stream")

6. 你知道哪些向量化模型？不同 embedding 模型怎么选？

答案：常见向量化模型可以分成通用语义模型、双语/多语模型、代码 embedding 模型和领域 embedding 模型。中文和中英混合场景常见的是 BGE 系列、text2vec、m3e、gte、Qwen embedding 等；英文或多语场景可以考虑 E5、GTE、BGE-M3；代码检索可以考虑 code embedding 模型；如果是电商、物流、医疗、金融这种强领域场景，还可以用领域数据做对比学习微调。

选型时不能只看公开榜单，要看自己的评测集。重点看 Recall@K、MRR、nDCG、跨语言召回、长文本截断、实体精确匹配、推理延迟和向量维度。比如跨境供应链里有中文问题、英文规则、商品编码和行业缩写，普通中文 embedding 可能不够，需要中英跨语义能力更强的模型。

embedding_candidates = {
    "bge-m3": ["multi-lingual", "dense+sparse", "long_context"],
    "bge-large-zh": ["chinese", "semantic_retrieval"],
    "gte-large": ["general_embedding", "english_chinese"],
    "text2vec": ["lightweight", "chinese"],
    "code-embedding": ["code_search", "api_search"]
}

7. 好的 Prompt 工程应该包含哪些内容？

答案：好的 Prompt 不只是把问题写清楚，而是要包含角色、任务目标、输入信息、输出格式、约束条件、可用工具、失败策略和示例。尤其在 RAG 和 Agent 场景里，Prompt 还要明确“只能基于证据回答”“证据不足时要说明不足”“不能伪造引用”“工具结果优先于模型记忆”。

对于结构化任务，最好要求模型输出 JSON，并给出 schema。对于高风险任务，要加入边界条件，比如不能生成法律最终结论，只能生成辅助审查建议。Prompt 工程的目标不是让模型更会说，而是让模型在可控边界内稳定完成任务。

prompt = """
你是跨境供应链合规审查助手。

任务：
根据用户问题、单据字段和检索证据，判断是否存在申报风险。

约束：
1. 只能基于【证据】回答；
2. 如果证据不足，输出“证据不足，无法判断”；
3. 不得编造规则编号；
4. 输出必须是 JSON。

输出格式：
{
  "has_risk": true/false,
  "risk_items": [
    {"field": "...", "reason": "...", "evidence_id": "..."}
  ],
  "suggestion": "..."
}
"""

8. RAG 的重排序有哪些方法？

答案：RAG 重排序常见方法有四类。第一类是基于规则的重排序，比如按文档版本、生效时间、权限、标题层级、字段匹配度加权。第二类是 Cross-Encoder rerank，把 query 和候选 chunk 拼接后打相关性分数，效果通常比纯向量更准。第三类是 LLM rerank，让大模型判断哪些证据最能回答问题，适合复杂多跳问题，但成本高、延迟大。第四类是学习排序，比如 LambdaMART 或基于点击/标注数据训练排序模型。

线上一般不会只用一种方法。常见做法是先用向量和关键词召回 Top100，再用 Cross-Encoder 排到 Top20，最后结合规则过滤和 MMR 去重选 Top5。这样可以兼顾速度、准确率和上下文多样性。

def rule_rerank(candidates):
    for c in candidates:
        score = c["semantic_score"]

        if c.get("is_latest_version"):