拼多多 AI Agent 开发一面

1、项目里负责的边界是什么？哪些内容是亲手实现或验证的？

我负责的边界一般分成四块：Agent 主链路、RAG 检索链路、工具调用层和效果评测。主链路包括 query 理解、路由、上下文构造、结果生成；RAG 这块包括文档清洗、切片、召回、重排和 prompt 拼接；工具调用层包括 function schema 定义、参数校验、执行封装和失败重试；评测这块包括 badcase 回流、离线测试集和线上日志分析。

亲手实现的部分主要是在线链路和评测闭环，比如 query 改写、混合检索、rerank 接入、上下文压缩、工具调用编排、token 统计、失败样本分析。验证过的内容包括不同 chunk 策略、不同 topK、不同 rerank 模型、不同 prompt 模板对最终回答质量的影响。

2、RAG 流程是什么？

RAG 一般分离线和在线两部分。离线侧先把文档接入，做清洗、去重、切片、embedding，然后写入向量库或混合检索库。在线侧用户问题进来后，先做 query 改写或者纠错，再做检索召回，取回候选片段后做 rerank，最后把问题和证据一起送给大模型生成答案。

如果是 Agent 场景，RAG 不是独立模块，而是和工具调用、记忆、状态管理一起协同的。比如用户问规则类问题就直接走检索，问实时数据就调工具，问混合问题就先查知识再结合工具结果生成答案。

def rag_pipeline(query):
    rewritten = rewrite_query(query)
    docs = retrieve(rewritten)
    docs = rerank(rewritten, docs)[:5]
    prompt = build_prompt(query, docs)
    return llm_generate(prompt)

3、实习项目里的 Agent 应用场景是什么？

4、多 Agent 协作机制怎么做？

多 Agent 协作一般不是让每个 Agent 随机聊天，而是做职责拆分。最常见的是一个 Planner 或 Router 负责拆解任务，再交给不同的执行 Agent。比如检索 Agent 负责查知识库，Tool Agent 负责调业务接口，Summary Agent 负责总结结果，Judge Agent 负责判断结果是否可用。

这样拆的好处是每个 Agent 的 prompt 更短，职责更清楚，工具也更少，稳定性比一个大一统 Agent 更高。最后由一个总控把各个 Agent 的输出汇总，生成最终回答。

class Planner:
    def plan(self, query):
        return ["retrieve", "tool", "summarize"]

class RetrieveAgent:
    def run(self, query):
        return ["活动规则片段", "类目限制片段"]

class ToolAgent:
    def run(self, query):
        return {"item_status": "已下架", "reason": "资质缺失"}

class SummaryAgent:
    def run(self, docs, tool_res):
        return f"根据规则和实时状态，当前商品{tool_res['item_status']}，原因是{tool_res['reason']}。"

def run(query):
    planner = Planner()
    steps = planner.plan(query)
    docs = RetrieveAgent().run(query) if "retrieve" in steps else []
    tool_res = ToolAgent().run(query) if "tool" in steps else {}
    return SummaryAgent().run(docs, tool_res)

5、记忆架构怎么做？

记忆一般分短期记忆、长期记忆和工作记忆。短期记忆保存最近几轮对话，用来保证多轮连续性；长期记忆保存用户偏好、稳定事实、长期设定；工作记忆保存当前任务的中间结果、已调过的工具结果和待确认状态。

实际系统里，不会把所有历史聊天原文一直带给模型，那样 token 会越来越大。通常做法是保留最近几轮原文，把更早的历史压缩成摘要，再根据当前 query 去召回相关长期记忆。这样既能保留上下文，又能控制成本。

class Memory:
    def __init__(self):
        self.recent = []
        self.long_term = {}

    def add_turn(self, msg):
        self.recent.append(msg)
        if len(self.recent) > 6:
            self.recent.pop(0)

    def save_fact(self, user_id, fact):
        self.long_term.setdefault(user_id, []).append(fact)

    def load_context(self, user_id):
        return {
            "recent": self.recent,
            "facts": self.long_term.get(user_id, [])
        }

6、RAG 效果怎么提升？如果出现幻觉怎么办？

RAG 的提升一般从四层做：文档质量、召回质量、排序质量、生成约束。文档先做清洗、去重、结构化切片，减少脏数据和重复数据；召回阶段通常会做 query rewrite、混合检索和 metadata 过滤；排序阶段加 rerank，把真正相关的 chunk 放到前面；生成阶段通过 prompt 约束模型必须基于证据回答。

幻觉通常有三种来源：没召回到、召回错了、模型不按证据答。没召回到就优化切片和召回；召回错了就优化重排和过滤；模型乱答就加明确限制和拒答机制。线上更稳的方式是把召回分数或证据置信度作为门槛，达不到阈值就直接拒答，而不是硬生成。

7、单个对话消耗的 token 大概多少？

没有固定值，要看系统 prompt 长度、历史轮数、检索上下文多少、工具描述多少、模型输出多长。普通单轮问答可能几百到一两千 token，带 RAG 后经常会上到两三千甚至五千以上。多轮对话如果带历史摘要、工具调用结果和多个检索片段，单次请求上万 token 也很常见。

Agent 场景的 token 消耗通常比普通聊天更高，因为还包含工具 schema、状态信息、记忆和中间结果。线上一般会记录 prompt token、completion token、总 token、平均耗时、不同场景 token 分布，再做成本分析。

8、单轮或多轮对话 token 比较大怎么办？

核心思路是裁剪、压缩、分流。单轮 token 大，通常是检索片段太多、工具描述太长、prompt 太啰嗦，可以减少 topK、先 rerank 再截断、把工具描述改成结构化 schema。多轮 token 大，通常是历史消息累积造成的，需要保留最近几轮原文，把更早历史压成摘要，长期偏好抽成结构化记忆。

另外还可以做模型分级，让 query 改写、分类、路由、摘要这类轻任务用小模型，大模型只处理最终生成。输出端也要限制 max_tokens，避免回答过长。