赛意信息 社招一面

社群小伙伴分享的社招面经 小伙伴说 基本上都是在问代码细节 , 我后来问他什么岗位没回 我去翻了翻官网招聘没翻到AI相关的岗位, 不知道在哪个平台投的

1. 自我介绍

2. 你们的工作流是怎么实现的？

答案：我们的工作流不是让大模型每次自由发挥，而是做成了状态机加任务编排。用户输入进来后，先经过意图识别，判断是知识问答、文档检索、代码分析、任务执行还是报告生成。然后进入 planner，planner 生成一个结构化计划，比如需要调用哪些工具、是否需要检索、是否需要用户补充信息。每一步执行后都会把结果写入状态，再由下一步读取状态继续执行。

工作流里最关键的是状态管理和失败恢复。比如一个任务需要先检索文档，再查接口，再生成测试用例，如果中间查接口失败，不能直接让整个任务崩掉，而是要记录失败原因，决定重试、降级还是返回部分结果。我们会给每次任务生成 run_id，所有模型调用、工具调用、检索结果和最终输出都挂在这个 ID 下，方便排查。

class WorkflowState:
    def __init__(self, query):
        self.query = query
        self.intent = None
        self.plan = []
        self.context = {}
        self.step_results = []
        self.status = "init"

def run_workflow(query, tools):
    state = WorkflowState(query)
    state.intent = classify_intent(query)

    if state.intent == "rag_qa":
        state.plan = ["retrieve", "rerank", "generate"]
    elif state.intent == "tool_task":
        state.plan = ["plan_tool", "execute_tool", "verify"]
    else:
        state.plan = ["generate"]

    for step in state.plan:
        try:
            result = tools[step](state)
            state.step_results.append({step: result})
        except Exception as e:
            state.status = "failed"
            state.context["error"] = str(e)
            break

    if state.status != "failed":
        state.status = "success"
    return state

3. LangChain 里接 MCP 客户端，需要做哪些处理后 Agent 才能用？

答案：MCP 客户端不能直接丢给 Agent 用，中间要做一层适配。MCP 服务暴露的是工具能力，Agent 需要的是带名称、描述、参数 schema、调用函数和返回格式的 Tool。适配层要完成几个事情：连接 MCP Server，拉取工具列表，把 MCP 的工具 schema 转成 LangChain Tool schema，封装异步调用，做参数校验、超时控制和异常处理。

这里最容易出问题的是参数。模型生成的工具参数不一定符合 MCP server 的要求，所以在 LangChain Tool 层要先做 pydantic 校验。其次 MCP 调用通常是异步的，而 Agent 执行链可能是同步或异步混合的，所以要处理好 async bridge。最后返回值也要标准化，否则 Agent 后面很难消费。

from langchain.tools import StructuredTool
from pydantic import BaseModel, Field
import asyncio

class SearchInput(BaseModel):
    query: str = Field(..., description="用户要检索的问题")
    top_k: int = Field(5, description="返回结果数量")

class MCPClient:
    async def call_tool(self, name, arguments):
        # 这里实际会通过 MCP 协议向 server 发请求
        return {"tool": name, "arguments": arguments, "result": "mock result"}

mcp_client = MCPClient()

def mcp_search_tool(query: str, top_k: int = 5):
    return asyncio.run(
        mcp_client.call_tool(
            "knowledge_search",
            {"query": query, "top_k": top_k}
        )
    )

search_tool = StructuredTool.from_function(
    func=mcp_search_tool,
    name="knowledge_search",
    description="用于检索企业知识库、接口文档、研发规范",
    args_schema=SearchInput
)

4. MCP 工具接入 Agent 时，怎么防止模型乱调工具？

答案：不能让模型看到工具就随便调。工具接入时要做白名单、权限校验、参数校验和风险分级。比如只读工具可以自动执行，写操作、删除操作、发布操作必须人工确认。Agent planner 生成动作后，不应该直接执行，而是先进入 action guard，由系统判断这个动作当前用户有没有权限、参数是否合法、是否属于高风险操作。

另外，工具描述也要写清楚。如果工具描述模糊，模型就容易误用。比如一个工具叫 execute 就很危险，模型不知道它到底执行什么；更好的命名是 query_api_doc、create_test_case_draft、search_error_log。工具名和描述本身就是 Agent 的约束。

def action_guard(user, action):
    readonly_tools = {"search_doc", "query_api", "retrieve_code"}
    high_risk_tools = {"delete_file", "publish_config", "run_shell"}

    if action["tool"] in readonly_tools:
        return True

    if action["tool"] in high_risk_tools:
        if not user.get("is_admin"):
            raise PermissionError("用户无权限执行高风险工具")
        if not action.get("human_confirmed"):
            raise PermissionError("高风险工具需要人工确认")

    return True

5. Milvus 向量检索脚本怎么写？从建表到检索说一下

答案：Milvus 的基本流程是连接服务、定义 collection schema、创建索引、插入向量和元数据、load collection、执行 search。真正项目里除了向量字段，还要保存 doc_id、chunk_id、title、content、source、version 这些元数据，否则检索出来没法回溯证据。

索引类型要根据数据规模选。小规模可以先用 FLAT 验证效果，大一点可以用 IVF_FLAT、HNSW。检索时还要带过滤条件，比如只查某个知识库、某个版本、某个权限范围内的文档。

from pymilvus import (
    connections, FieldSchema, CollectionSchema,
    DataType, Collection
)
import numpy as np

connections.connect(alias="default", host="localhost", port="19530")

dim = 768

fields = [
    FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
    FieldSchema(name="doc_id", dtype=DataType.VARCHAR, max_length=128),
    FieldSchema(name="chunk_id", dtype=DataType.VARCHAR, max_length=128),
    FieldSchema(name="content", dtype=DataType.VARCHAR, max_length=4096),
    FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=dim),
]

schema = CollectionSchema(fields, description="研发知识库向量表")
collection = Collection(name="dev_knowledge_base", schema=schema)

index_params = {
    "index_type": "HNSW",
    "metric_type": "COSINE",
    "params": {"M": 16, "efConstruction": 200}
}

collection.create_index(
    field_name="embedding",
    index_params=index_params
)

vectors = np.random.random((3, dim)).astype("float32").tolist()

collection.insert([
    ["doc_001", "doc_002", "doc_003"],
    ["chunk_001", "chunk_002", "chunk_003"],
    ["接口鉴权说明", "异常码定义", "测试规范说明"],
    vectors
])

collection.flush()
collection.load()

query_vec = np.random.random((1, dim)).astype("float32").tolist()

res = collection.search(
    data=query_vec,
    anns_field="embedding",
    param={"metric_type": "COSINE", "params": {"ef": 64}},
    limit=5,
    output_fields=["doc_id", "chunk_id", "content"]
)

pri