大厂智能体岗位的面试已经不再是“你调过几个大模型 API”这么简单。最近无论是搜 deepseek harness、codex harness还是搜 Multi-Agent 开发、A2A 智能体协作其实都在补同一块内容如何把单个 Agent 从“能跑通的 Demo”变成稳定、可扩展、可运维的服务以及当多个 Agent 一起协作时怎么通信、怎么分发任务、怎么处理高并发。本篇文章围绕 Harness 架构、Multi-Agent、A2A 协议、Sub-agent 四个关键词拆成两个可落地的项目来精讲项目 A 把一个裸 Agent 改造成带 Harness 的可运维服务项目 B 基于 A2A 协议实现一个 Multi-Agent 工单协作平台并讨论高并发场景下的异步化、幂等、限流和监控。所有代码和配置都按最小可运行示例给出正式落地时再结合你自己的模型接入方式、消息队列选型和部署环境调整。1. 先把四个关键概念对齐后面讲项目才不会跑偏这四个词经常在一个 JD 里同时出现但它们解决的是不同层次的问题。如果面试回答时把 Harness 和 Agent、A2A 和 Multi-Agent 混在一起讲项目听起来会很散。先各自对齐。1.1 Agent 是决策者Harness 是运行外壳两者分工不同一个 Agent 最核心的部分是循环接收用户输入判断是否需要调用工具拼接上下文再调用大模型最后返回结果。它真正要解决的问题是“下一步该做什么”。但一个只包含这个大模型的循环离生产可用还很远。Harness 在智能体工程里指的就是围绕 Agent 循环搭起来的那一层“运行外壳”。它负责工具注册、参数校验、会话管理、上下文预算、超时重试、错误恢复、日志和可观测性、安全边界等。Agent 负责判断Harness 负责保障判断能稳定被执行。维度Agent 核心Agent Harness主要职责推理、规划、调用工具让 Agent 在真实环境稳定运行核心问题该做什么怎么做完、做错怎么回退、崩溃怎么恢复典型代码Prompt、LLM 调用、工具结果拼接工具注册表、生命周期管理、Observer、沙箱出错影响答案不准进程卡死、状态错乱、资源泄漏、请求雪崩面试常问怎么设计 ReAct 循环怎么控制上下文、怎么限流、怎么观测很多热门搜索词比如 deepseek harness、codex harness本质上反映的是同一件事开发者发现光有大模型 API 不够必须给特定模型套一个工程外壳。不同项目里的 harness 实现方式不同有的是一个启动器有的是一套 Agent 生命周期框架有的是一个调用大模型的封装层但目标一致让模型在受控、可观测、可恢复的边界内工作。一个最小裸 Agent 可能长这样# bare_agent.py 仅用于说明问题不是生产实现 def run_agent(user_input, history): tools {calculator: calculator, search: search} messages history [{role: user, content: user_input}] while True: response llm.chat(messages, toolstools) if response.tool_calls: for call in response.tool_calls: tool_name call.name result tools[tool_name](**call.arguments) messages.append({ role: tool, tool_call_id: call.id, content: result }) continue return response.content这段代码的问题很明显没有判断工具是否存在时直接取值没有超时控制历史消息无限增长也没有任何日志和监控。坏一个工具调用整个请求就崩了。这就是 Harness 要解决的第一个问题给 Agent 加上边界和保障。1.2 Multi-Agent 不是“多个 Agent 排排站”而是靠 Sub-agent 拆解分工当单个 Agent 的任务变复杂后会出现几个实际瓶颈上下文太长导致模型忽略关键信息工具太多导致 Prompt 指令容易被稀释多个任务混在一起时很难保证执行顺序。于是出现了一种常见的组织方式主 Agent 负责接收请求、拆解任务然后动态委派给不同的 Sub-agent 去处理最后把结果汇总。Sub-agent 在这里扮演的是“专项执行者”。比如一个智能客服系统里主 Agent 判断用户是投诉、退款还是技术咨询然后分别交给投诉组 Sub-agent、财务组 Sub-agent、技术支持 Sub-agent。每个 Sub-agent 的 Prompt、工具和上下文都会更聚焦出错概率更低也更容易单独扩容和部署。Multi-Agent 不意味着越多越好。实际工程里每多一个 Sub-agent就多一套部署、超时、重试、权限和观测成本。合理的设计是主 Agent 只做拆解和汇总Sub-agent 只做单一职责节点数量控制在能覆盖业务场景的最小范围。1.3 A2A 协议解决的是 Agent 和 Agent 之间怎么通信MCP 已经解决了一个问题Agent 如何稳定地调用外部工具。但 Agent 之间如何发现彼此、如何派发任务、如何把结果传回来是另一个层次的问题这就是 A2A 这类 Agent 通信协议的价值。A2A 是面向 Agent 间协作的开放通信协议常见实现会围绕几个核心概念展开Agent Card 描述 Agent 的能力和接入地址Task 表示一次任务生命周期Message 表示任务中的消息Artifact 表示任务产生的文件、结构化数据或最终结果。面试中不用背协议版本但要理解这些概念为什么要出现。维度MCPA2A通信对象Agent 与工具/服务Agent 与 Agent解决的核心问题工具接入标准化任务发现、分配、结果回传标准化常见元素Resource、Tool、PromptAgent Card、Task、Message、Artifact典型场景数据库查询、外部 API 调用主 Agent 派活给 Sub-agent子任务状态同步需要留意的是A2A 这类协议还处于快速演进阶段不同框架对协议的支持成熟度不一样。落地方案不要只看配置名称要确认框架底层是否真的实现了 Agent Card 注册、Task 状态流转和结果回传。能跑通 Demo 不代表协议语义完整实现。1.4 一句话把四个概念串起来可以把这套体系理解为一家公司Agent 是员工负责具体判断和行动Harness 是办公环境和管理制度保证员工在安全边界内干活Sub-agent 是负责专项工作的专员A2A 是部门之间的标准沟通协议Multi-Agent 是整套组织架构。这个类比很适合在面试开头使用。它能让面试官立刻知道你对这四个概念的理解不是背名词而是从工程协作的角度去理解的。2. 项目 A给一个裸 Agent 套上 Harness把它变成可运维服务项目 A 的目标很明确把一个只能在本机 while True 循环里跑通的裸 Agent改造成一个带工具注册表、会话隔离、超时重试、日志观测的 HTTP 服务。这个项目做完后你可以回答“Harness 是什么”“Harness 里到底写了什么代码”这类问题。2.1 从上一段的裸 Agent 开始看清问题清单裸 Agent 最常见的几个问题可以列成一张表。这张表其实可以直接变成项目 A 的需求清单。问题现象后果工具不存在时直接取 dict 值KeyError请求中断LLM 调用没有超时网络慢时请求一直挂着连接和内存资源被占满历史消息无限追加Prompt 越来越长Token 费用上涨、模型回答质量下降没有会话隔离不同用户共享同一段上下文用户数据串线没有日志和 trace问题无法复现生产排障只能靠猜工具副作用不可控工具重复调用或调用失败订单重复、账号原地等异常项目 A 的 Harness 设计就围绕这些问题逐项解决。2.2 Harness 工程目录设计在常见项目里目录可以按下面这种结构组织agent_harness/ main.py configs/ agent.yaml harness/ __init__.py loop.py registry.py session.py observer.py memory.py tools/ __init__.py base.py calculator.py search.py tests/ test_registry.py test_loop.py目录结构不是标准答案但每个模块的职责要单一configs/agent.yaml存放模型名、超时、历史最大轮数、温度等配置。harness/registry.py工具注册表统一管理工具名称、描述、参数 Schema。harness/loop.pyAgent 主循环负责调用模型、判断工具调用、执行工具、处理异常。harness/session.py会话管理和上下文预算。harness/memory.py记忆裁剪或摘要。harness/observer.py日志、事件回调用于可观测性。harness/tools/具体工具实现。先看配置文件# configs/agent.yaml model: name: your-model-name temperature: 0.3 max_tokens: 2048 harness: max_steps: 10 llm_timeout_seconds: 60 tool_timeout_seconds: 15 max_history_rounds: 20 session_ttl_seconds: 1800 tools: calculator: enabled: true search: enabled: true这里的max_steps很关键。它限制 Agent 一轮请求内最多执行多少次工具调用防止模型进入死循环也防止恶意输入诱导模型反复调用工具最终拖垮下游服务。2.3 核心实现工具注册表、主循环、HTTP 暴露工具注册表是 Harness 里最容易理解、也最容易被忽略的一部分。它不只是把工具放到 dict 里还要做名称冲突校验、参数 Schema 校验和职责声明。# harness/registry.py from dataclasses import dataclass from typing import Callable, Any, Optional dataclass class ToolSpec: name: str description: str parameters: dict func: Callable[..., Any] timeout_seconds: int 15 class ToolRegistry: def __init__(self): self._tools {} def register(self, spec: ToolSpec): if spec.name in self._tools: raise ValueError(ftool name duplicated: {spec.name}) if not spec.parameters: raise ValueError(ftool must declare parameters schema: {spec.name}) self._tools[spec.name] spec def get(self, name: str) - Optional[ToolSpec]: return self._tools.get(name) def list_tools(self) - list: return [ {name: s.name, description: s.description, parameters: s.parameters} for s in self._tools.values() ]注册工具时强制要求参数 Schema 存在这能避免裸 Agent 里“传入一个无法解析的参数”的问题。大模型生成的 tool_call 参数是字符串Harness 要做一次 JSON 解析和校验而不是直接把字符串塞给函数。主循环部分要做的事比较多# harness/loop.py import json import time import traceback class HarnessLoop: def __init__(self, registry, session, observer, config): self.registry registry self.session session self.observer observer self.config config def run(self, user_input: str): session self.session.get_or_create() messages session.load_messages() messages.append({role: user, content: user_input}) for step in range(self.config[harness][max_steps]): deadline time.time() self.config[harness][llm_timeout_seconds] try: response self._call_llm_with_timeout(messages, deadline) except TimeoutError: self.observer.record(llm_timeout, stepstep) raise except Exception as exc: self.observer.record(llm_error, excstr(exc)) raise self.observer.record(llm_response, stepstep, contentresponse.content) if not response.tool_calls: session.save_messages(messages [{role: assistant, content: response.content}]) return response.content for call in response.tool_calls: messages.append({ role: assistant, tool_calls: [{id: call.id, name: call.name, arguments: call.arguments}] }) result self._execute_tool(call) messages.append({ role: tool, tool_call_id: call.id, content: result }) raise RuntimeError(max steps reached)这个实现和裸 Agent 最大的区别是每一步都有超时、每次工具调用都走注册表、每一步都写 Observer 日志、超过最大步骤数会主动中断。否则模型一旦反复调用工具请求会无限挂起。再通过 FastAPI 把服务暴露出来# main.py from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class ChatRequest(BaseModel): session_id: str message: str app.post(/chat) def chat(req: ChatRequest): session_id req.session_id user_input req.message harness build_harness(session_id) result harness.run(user_input) return {session_id: session_id, reply: result} app.get(/health) def health(): return {status: ok}到这里一个裸 Agent 已经变成了一个可以被 HTTP 调用、有会话隔离、有工具注册、有超时限制的服务。2.4 运行验证不只是看能不能返回结果启动服务后先用健康检查确认基础状态curl http://127.0.0.1:8000/health然后用一个需要调用工具的请求验证 Harness 闭环curl -X POST http://127.0.0.1:8000/chat \ -H Content-Type: application/json \ -d {session_id: s1, message: 请计算 17 * 23}预期日志里会出现两次关键记录eventllm_response step0 content我将调用计算器工具 eventtool_execute toolcalculator args{expression: 17*23} eventllm_response step1 content17 * 23 的结果是 391只看到返回结果还不够要确认首次调用时工具参数是否正确解析。二次调用时模型是否把工具结果写进最终回复。如果故意让工具抛异常服务是否返回错误而不是进程崩溃。项目 A 最容易踩的坑有三个。第一个把工具直接写死在主循环里没有注册表导致工具一多就乱第二个历史消息无限增长要加max_history_rounds或摘要策略第三个LLM 调用没有超时生产环境一抖动整个服务就挂掉。Harness 的核心价值就是让这三个问题在框架层面被解决而不是每次靠开发人员写临时补丁。3. 项目 B用 A2A 协议把 Multi-Agent 和 Sub-agent 串起来项目 A 解决单个 Agent 的稳定性之后项目 B 进入协作阶段。典型的场景是工单分发用户提交一条工单主 Agent 先判断工单类型再分发给不同的 Sub-agent 处理。3.1 业务场景与总体目标假设平台每天会收到几千条工单类型包括售后、技术支持、财务退款。主 Agent 负责理解用户描述拆解工单类型再交给对应的 Sub-agent。每个 Sub-agent 本身也是一个带 Harness 的 Agent 服务可以复用项目 A 的工程结构。项目 B 要解决的不只是“把 A2A 请求发出去”还包括Sub-agent 的能力如何被发现即 Agent Card 注册。任务如何被创建Task 生命周期如何流转。高并发下如何不把主 Agent 拖垮。同一个任务重试时如何避免重复处理。目录结构可以这样组织a2a_platform/ a2a/ agent_card.py message.py task.py transport.py agents/ orchestrator.py support_agent.py tech_agent.py workers/ dispatcher.py store/ task_store.py configs/ a2a.yaml tests/a2a/是协议适配层agents/是具体 Agent 逻辑workers/是异步派发消费者store/保存任务状态。3.2 A2A 任务消息的最小结构在项目 B 里不依赖任何厂商的完整协议栈自己定义一个最小 A2A 消息结构方便理解核心机制{ id: task_1001, type: task/create, agent_card: { name: support_agent, url: http://support-agent:8001/a2a, capabilities: [refund, after_sale] }, message: { role: user, parts: [ {kind: text, text: 用户申请退款订单号 20261101} ] }, metadata: { trace_id: trace_88a1, source: orchestrator } }Task 状态机在常见实现中至少要有四个状态submitted、working、completed、failed。可以做一张表状态含义谁来触发submitted任务已创建并进入队列orchestratorworkingSub-agent 已接收并开始处理worker / sub-agentcompleted处理成功结果已回传sub-agentfailed处理失败留下错误信息sub-agent / worker 超时状态机最重要的一点是不允许从 completed 或 failed 直接跳回 working。这个约束能避免重复消费或乱序更新。3.3 主 Agent 编排与 Sub-agent 注册发现Sub-agent 通过 Agent Card 暴露自己的能力和接入地址。主 Agent 在派发前先根据能力匹配注册表# agents/registry.py class AgentRegistry: def __init__(self): self._cards {} def register(self, card: dict): self._cards[card[name]] card def find_by_capability(self, capability: str): return [card for card in self._cards.values() if capability in card.get(capabilities, [])]主 Agent 的编排逻辑只需要做三件事拆解任务选择 Sub-agent创建并派发 Task。# agents/orchestrator.py class Orchestrator: def __init__(self, registry, task_store, dispatcher): self.registry registry self.task_store task_store self.dispatcher dispatcher def handle(self, user_request: str): category self.detect_category(user_request) candidates self.registry.find_by_capability(category) if not candidates: return {error: fno agent for category {category}} task self.create_task(candidates[0], user_request) self.task_store.save(task) self.dispatcher.dispatch(task) return {task_id: task[id], status: submitted}这里的detect_category可以是 LLM 分类器也可以是关键词规则。项目 B 里建议先用规则跑通再换成 LLM因为这样你能对照两种方式的准确率和延迟。3.4 高并发落地异步任务队列而不是同步等待如果不做异步化主 Agent 直接同步请求 Sub-agent会暴露三个问题Sub-agent 响应慢主 Agent 的线程被长期占用。瞬时流量到来时所有线程都会卡在外部 HTTP 请求上。Sub-agent 扩容或重启后URL 会变化同步调用方必须跟着改配置。所以项目 B 引入消息队列或 Redis Stream把“派发”动作和“执行”动作解耦。# workers/dispatcher.py class Dispatcher: def __init__(self, queue, task_store): self.queue queue self.task_store task_store def dispatch(self, task): # 写入队列前先保存任务保证不丢 self.task_store.update_status(task[id], submitted) self.queue.xadd(agent_tasks, task)消费者 worker 从队列里拿到任务后先做幂等检查再调用 Sub-agent 的 A2A endpoint# workers/consumer.py def process_message(payload): task_id payload[id] if not task_store.acquire_lock(task_id): return # 已处理或处理中 try: task_store.update_status(task_id, working) result a2a_transport.post_task(payload) task_store.update_status(task_id, completed, resultresult) except Exception as exc: task_store.update_status(task_id, failed, errorstr(exc))这里最关键的是acquire_lock。队列在极端情况下会重复投递如果 Sub-agent 不是幂等的退款、通知、发货这类操作会被执行两次。锁的粒度必须是 Task id。3.5 验证方法最小验证分三步启动三个 Sub-agent 服务各自注册 Agent Card。启动主 Agent 服务提交不同类型工单。观察 task_store 中 Task 状态是否依次从 submitted 到 working 到 completed。如果为了模拟高并发可以写一个脚本并发提交 1000 个工单python load_test.py --count 1000 --concurrency 50然后观察队列深度、平均延迟、failed 数量和日志中是否有重复 task_id。失败的工单不应该静默丢弃至少要进入失败队列或标记为 failed 并设置可重试次数。项目 B 的常见坑也很集中任务重复消费、状态更新乱序、Sub-agent 超时导致任务永远停在 working。解决办法分别是 Task id 幂等锁、状态机校验、任务级超时和租约机制。4. Multi-Agent 高并发落地并发模型、超时、幂等、限流、监控项目 B 已经把基础闭环跑通但要拿到大厂岗位的面试认可还需要把“高并发怎么设计”讲清楚。这一层是最容易被量化考核的部分。4.1 并发模型选型要看业务不能只看技术热度方案适合场景优点缺点同步 HTTP 调用并发量低、依赖稳定、要求立即返回实现简单排障直观调用方线程被占用雪崩风险高asyncio 异步IO 密集Sub-agent 数量少单线程处理大量请求资源占用低不适合耗时过长任务取消和超时复杂线程池/进程池CPU 密集或调用方有并发上限隔离不同任务简单连接池和队列参数需要调优消息队列 Worker跨服务、削峰填谷、需要重试解耦强适合大流量引入消息队列延迟变高需处理重复消费项目 B 的高并发设计通常选消息队列 Worker。因为工单处理不要求毫秒级实时返回系统需要的是“先把任务接下再慢慢消化”。队列天然是削峰填谷的工具。4.2 关键参数要能解释“为什么这么配”面试时一个很高分的回答不是“我配了 timeout”而是“我把 timeout 设为 60 秒是因为 Sub-agent 平均 P99 延迟是 8 秒重试 3 次留给单次 LLM 调用的上限是 60 秒”。参数常见默认值调大的影响调小的影响推荐思路max_workers10吞吐上升但下游压力变大吞吐下降队列积压结合下游 QPS 上限压测llm_timeout60s更容忍模型慢更容易误杀取模型响应 P99 再加余量tool_timeout15s容忍第三方 API 慢工具容易失败按调用工具的真实耗时设置retry 次数3成功率更高但重复风险大成功率下降重试必须配幂等和退避queue_depth取决于队列上限容灾能力强但延迟变大容易丢任务设置告警线如 80%rate_limit按模型 RPM压垮模型服务任务积压看模型供应商配额参数表要结合你真实验证的数据不要背默认值。项目里所有参数都应该有压测或日志作为支撑这才是“真实落地”和“抄配置”的区别。4.3 可观测性是 Multi-Agent 项目里最容易漏的部分当任务横跨主 Agent、消息队列、Sub-agent 三层之后排查一个失败的工单会变得很困难。解决方案是从入口开始生成 trace_id并把它写进 Task metadata、队列消息和日志中。{ts: 2026-01-01T12:00:00Z, trace_id: trace_88a1, event: dispatch_task, task_id: task_1001, status: submitted}建议至少记录三类指标任务延迟从 submitted 到 completed 的耗时。失败率failed / submitted 的占比。队列深度未消费消息数量。一旦队列深度持续上涨说明 Sub-agent 消费能力不足需要扩容或检查下游模型 API 是否被限流。没有了 trace 体系这类问题排查会非常低效。4.4 权限和安全边界要让每个 Sub-agent 尽量“少拿权限”Multi-Agent 系统最大的安全隐患是主 Agent 一旦被绕过所有 Sub-agent 能力都会暴露。项目 B 在设计时每个 Sub-agent 只注册自己需要的最小工具集。比如财务 Sub-agent 不必要有修改工单备注的权限技术 Sub-agent 不应该能访问退款接口。A2A endpoint 也不要直接暴露公网保持在内部网络并加身份校验和调用方白名单。5. 双项目怎么变成大厂面试里的项目亮点技术做完后还要解决“怎么讲”的问题。很多候选人项目确实做了但讲得像个功能列表面试官听完记不住任何取舍。5.1 面试官想听的是边界、取舍和故障处理同一个项目两种讲法差异很大。普通版本“我做了智能体项目支持多智能体协作用了 A2A 协议支持高并发。”推荐版本“我给单个 Agent 建了 Harness解决了工具扩展和上下文失控的问题。然后在工单场景里引入 Sub-agent通过 A2A 协议做任务分发。第一版是同步调用压测时发现 Sub-agent 一慢整个服务就阻塞所以改成消息队列异步化同时设计了 Task id 幂等锁和状态机任务重复消费的问题通过锁解决了。线上观察的核心指标是队列深度、任务延迟、失败率收到过 Sub-agent 超时告警处理方式是加了租约机制超时任务能被其他 worker 接管。”后者听上去“真实”因为有明确的失败和取舍。5.2 项目 A 的讲述模板描述项目 A 时重点放在裸 Agent 有哪些问题。Harness 拆成哪些模块。怎么保证工具注册和参数校验。怎么防止上下文无限增长。超时和重试是怎么设计的。为什么不用现成框架或者为什么选择自研部分模块。如果面试官问“这种能力不是可以直接用 LangChain 或 AgentScope 吗”不要急着否定框架。可以回答现成框架能覆盖大部分场景但自己实现 Harness 是为了理解核心机制同时保留对工具副作用、上下文预算和观测体系的自控能力。生产环境完全可以基于成熟框架二次开发关键是判断标准要清楚。5.3 项目 B 的讲述模板项目 B 的重点是任务分发和状态一致性。讲述顺序业务场景工单分类派发。协议层使用 A2A 风格的最小协议包含 Agent Card、Task、Message。编排层主 Agent 只负责拆解和路由。异步化写队列、消费、调用 Sub-agent。一致性幂等锁、状态机、超时租约。验证并发提交、队列监控、故障注入。不要强调“我用了某某张架构图”而是要强调“任务在哪里被阻塞失败后怎么恢复重复消息会不会造成二次操作”。5.4 面试技术追问自查表追问问题你需要能答出的点Harness 和框架有什么区别Harness 是工程外壳框架是完整开发套件边界看项目需要Sub-agent 怎么动态发现Agent Card 注册、能力匹配、注册表心跳或配置刷新同一个任务重复投递怎么办Task id 做幂等锁处理中/已完成直接跳过Sub-agent 超时但实际还在跑怎么办用 lease 时间超时任务可被其他 worker 接管队列消息丢失怎么办先写库再投递消费后确认失败进重试队列怎么控制成本上下文预算、模型分级、Sub-agent 只调用必要模型多个 Sub-agent 结果怎么合并由主 Agent 收集 Artifact 和最终 message再做摘要5.5 简历关键词和项目描述写简历时关键词不要堆“精通、架构”这类词而要把结果写进去基于 Harness 设计单 Agent 运行框架完成工具注册、上下文预算、超时重试和日志观测支撑日请求量 XXX。基于 A2A 协议实现 Multi-Agent 工单协作引入消息队列异步化通过 Task 状态机和幂等锁解决重复消费问题压测支撑 XXX TPS。数字必须是真实压测或线上数据。如果没有不要编造。面试官连续追问数据来源时编造的线上数据很容易穿帮。6. 常见问题排查从现象倒推根因项目 B 这类系统问题往往横跨多个组件。排错时要按链路一层层查而不是看到报错就改代码。6.1 错误现象速查表现象可能原因检查方式处理建议工单一直 submitted消息队列积压或消费者没启动看队列深度、消费组 lag扩容 worker检查消费者启动日志工单状态是 working 但很久没结果Sub-agent 超时或崩溃看 Sub-agent 日志、task lease 时间加超时租约超时任务重新派发同一条工单被处理两次消费者重复消费幂等锁未生效查 task_id 日志出现次数用 Redis setnx 或数据库唯一约束调用 LLM 经常报超时模型 API 慢或触达限流看模型服务端响应码和耗时调整 timeout、增加退避重试Sub-agent 找不到Agent Card 未注册或注册表过期检查 AgentCard 注册接口增加注册表刷新或心跳任务失败但无任何日志异常被吞掉检查代码中是否有裸 except统一日志组件至少记录 trace_id6.2 排查链路先别改代码按顺序确认推荐顺序是确认输入和 task_id问题工单是否存在携带的 trace_id 是什么。确认路由主 Agent 是否把任务分给了正确的 Sub-agent。确认状态机Task 当前状态是否合法是否存在乱序更新。确认队列任务是否被消费消费组是否有积压或重复消费。确认 Sub-agentSub-agent 是否收到了请求调用是否超时。确认模型 APILLM 调用是否稳定响应延迟是否升高。确认资源CPU、内存、连接池、Redis 连接数是否正常。不要一开始就去翻 Sub-agent 代码。很多问题在队列层或者注册层就能定位直接查代码反而浪费时间。6.3 日志关键字要齐全项目上线前至少保证以下日志可查eventtask_createdeventtask_dispatchedeventtask_receivedeventtask_completedeventtask_failedeventllm_timeouteventretry_exceededeventidempotency_skip只要每一条日志都带 trace_id 和 task_id排查链路就能很顺畅。否则任何一次线上故障都可能变成一次大型考古。7. 备战 2026可复用的工程落地清单和练习路径最后给你两份可以直接用于开发前或面试前自检的清单。7.1 Harness 落地检查清单工具是否通过注册表管理禁止在 Agent 循环里硬编码工具。每个工具是否声明参数 Schema并在执行前做 JSON 解析和校验。LLM 调用是否有超时超时后是否有明确异常分支。工具调用是否有独立超时是否与 LLM 超时分开。历史消息是否有最大轮数或 token 预算。不同会话是否隔离session key 是否会过期。每一步工具调用是否有日志和 trace。达到 max_steps 后是否主动中断并返回可读错误。生产环境是否有限流防止单个用户刷爆模型 API。工具是否有副作用控制同一个副作用操作是否幂等。7.2 Multi-Agent 高并发检查清单每个任务是否有全局唯一 task_id。Task 状态机是否有明确流转规则。消费者是否处理重复消息Task id 幂等锁是否生效。队列消息是否先落库再投递。Sub-agent 是否有超时和租约机制。重试是否采用退避策略是否避免重试风暴。链路是否有 trace_id。是否有队列深度、任务延迟、失败率三类指标。每个 Sub-agent 是否只开放最小权限。是否对模型 API 做限流保护。7.3 练习路径建议如果直接从零开始准备建议按下面顺序练用一周时间做项目 A把裸 Agent 改成 Harness 服务。给项目 A 加上流式输出和 SSE体验长响应下的连接管理。基于项目 A 做两个业务 Agent再进入项目 B。项目 B 第一版用同步 HTTP压测发现问题后再改成消息队列异步化。引入 A2A 风格的最小协议定义 Agent Card 和 Task 状态。加入幂等、状态机、超时租约做故障注入验证。这套路线最大的好处是每一层都有明确的前置依赖没有单个 Agent 的稳定性Multi-Agent 协作就是建立在沙地上没有任务状态机高并发下一定会出现状态混乱没有幂等重试机制一旦开启就会带来业务事故。把单个 Agent 的 Harness 写稳再谈 A2A 和 Multi-Agent是备战 2026 智能体岗位最稳妥的路径。