AI Agent 技术月度展望：总结 7 月的收获并规划 8 月的学习与实践方向

一、深度引言与场景痛点

7 月收官了。10 篇文章写完，回头看这个月在 Agent 开发上走过的路，既有成就感也有遗憾。

成就感是实打实的：Agent 流水线从裸奔到分层架构上线，RAG 系统延迟从 5 秒压到 200ms，向量检索从单编码切换到混合检索，Prompt 工程从靠直觉变成了靠量化测试。这些都是这个月的硬产出。

遗憾同样真实：Agent 的并行编排还没实现，依然在串行跑；RAG 的检索精度虽然提升了不少，但在长文档场景仍然会丢失关键细节；向量检索前沿追踪了一堆，真正落地的只有 Scalar 量化。计划做的事比做完的事多——这是每个月的常态，但 7 月尤其明显。

展望 8 月，最迫切的痛点有三个：

并行编排的空白。7 月的 Agent 是串行流水线（路由→记忆→规划→执行→审计），每个环节都在等前一个完成。但实际业务中，很多工具调用是独立的——查天气和查新闻可以同时进行，不需要等查天气完了才查新闻。串行编排白白浪费了时间，P99 延迟被拖到了 500ms+。

多 Agent 协作的低效。三个 Agent（意图识别、执行、总结）之间通过硬编码的 pipeline 串联，没有动态分工。意图识别 Agent 在简单查询上是多余的——直接调用执行 Agent 就够了，但当前架构不允许跳步。需要一个更灵活的"编排引擎"，根据任务复杂度动态决定走哪条路径。

RAG 检索精度的天花板。混合检索把精度从 72% 提到 89%，但剩下的 11% 的失败案例几乎都是"长文档中的细节信息被截断丢失"。智能截断策略虽然减少了 Token 数量，但也丢掉了部分关键段落。需要一个更精细的"段落级检索"方案。

下面这张图梳理了 7 月的成果闭环和 8 月的规划方向：

二、底层机制与原理深度剖析

8 月的三个核心规划方向，背后的原理分别对应 Agent 系统的三个进化维度：

DAG 并行编排——从串行流水线到 DAG 执行引擎。串行编排的问题是显而易见的：每一步都要等前一步完成。但大部分 Agent 工具调用之间存在"可并行"的独立关系。比如一个旅行规划 Agent，用户问"明天北京天气怎么样、有什么好吃的餐厅"，查天气和查餐厅是完全独立的两个工具调用，不需要串行等待。

DAG 编排引擎的核心是：将任务分解为 DAG（有向无环图），DAG 中的节点是任务步骤，边是依赖关系。无依赖的节点可以并行执行，有依赖的节点必须等前置节点完成。执行引擎按照拓扑排序调度任务，每个"层"内的节点全部并行，不同"层"之间的节点串行等待。这样既保证了依赖关系的正确性，又最大化了并行度。

动态路由引擎——从固定流水线到自适应编排。当前架构是固定的 Router→Memory→Planner→Executor→Auditor 五步流水线。但实际场景中，简单查询（"北京今天天气"）只需要 Router→Executor 两步就够了，不需要 Planner 拆解、不需要 Auditor 校验。复杂查询（"帮我规划下周的出差行程，考虑天气、交通、酒店预订"）才需要完整的五步流水线。

动态路由引擎的原理是：在 Router 之后加一个"复杂度评估器"，根据任务需要的工具数量、推理步骤数、上下文依赖深度等指标，动态选择执行路径。简单任务走"快速路径"（Router→Executor），复杂任务走"完整路径"（Router→Memory→Planner→Executor→Auditor）。这样简单查询的延迟可以控制在 200ms 以内，复杂查询走完整流程保证质量。

段落级检索——从文档级到 sentence-level 的精度跃升。当前 RAG 系统的检索粒度是"文档级"——每条向量对应一篇完整文档。检索 Top-K 返回的是 K 篇文档，然后在文档内做截断。这种方式的问题是：一篇 5000 字的文档可能只有一小段与用户查询相关，但检索时整篇文档作为一个向量，语义被平均化了——相关段落和不相关段落混在一起，向量表示不够精确。

段落级检索的原理是：把每篇文档按 sentence 或 paragraph 拆分，每个段落单独生成 embedding 和向量索引。检索时直接匹配段落级向量，返回的 Top-K 是 K 个最相关的段落，而不是 K 篇文档。这样检索精度从文档级的 89% 可以提升到段落级的 95% 以上。代价是向量数量增加了（一篇 100 段的文档变成 100 个向量），但通过 metadata 关联（段落→文档→来源）可以轻松回溯完整上下文。

三、生产级代码实现

以下是一个 DAG 并行编排引擎的实现，整合了动态路由和自适应路径选择：

import asyncio import json import time from dataclasses import dataclass, field from enum import Enum from typing import Any import structlog logger = structlog.get_logger() # ========== DAG 任务模型 ========== class TaskStatus(Enum): PENDING = "pending" RUNNING = "running" SUCCESS = "success" FAILED = "failed" SKIPPED = "skipped" class TaskComplexity(Enum): SIMPLE = "simple" # 1-2 步，无需规划 MEDIUM = "medium" # 3-5 步，需要规划 COMPLEX = "complex" # 6+ 步，需要完整流水线 @dataclass class DAGTask: """DAG 中的单个任务节点。""" task_id: str name: str handler_name: str # 对应的处理函数名 args: dict[str, Any] = field(default_factory=dict) depends_on: list[str] = field(default_factory=list) # 依赖的任务 ID status: TaskStatus = TaskStatus.PENDING result: Any = None error: str | None = None latency_ms: float = 0.0 retry_count: int = 0 max_retries: int = 2 # ========== 动态路由引擎 ========== class DynamicRouter: """动态路由引擎：根据任务复杂度选择执行路径。 7 月痛点：固定五步流水线对简单查询浪费步骤。 8 月方案：复杂度评估 → 自适应路径选择。 """ def evaluate_complexity(self, query: str) -> TaskComplexity: """评估任务复杂度。""" # 规则 1：工具数量推断 tool_indicators = [ "查", "搜索", "找", # 单工具 "帮我", "规划", "对比", "分析", # 多工具 "同时", "还要", "并且", # 并行工具 ] single_tool_count = sum( 1 for kw in tool_indicators[:3] if kw in query ) multi_tool_count = sum( 1 for kw in tool_indicators[3:6] if kw in query ) parallel_count = sum( 1 for kw in tool_indicators[6:] if kw in query ) # 规则 2：推理深度推断 reasoning_indicators = [ "为什么", "原因", "如何", # 需要推理 "评估", "优化", "建议", # 需要深度推理 ] reasoning_count = sum( 1 for kw in reasoning_indicators if kw in query ) # 规则 3：查询长度推断 query_length = len(query.split()) # 综合评分 complexity_score = ( multi_tool_count * 2 + parallel_count * 3 + reasoning_count * 2 + min(query_length / 5, 3) ) if complexity_score <= 2: return TaskComplexity.SIMPLE elif complexity_score <= 6: return TaskComplexity.MEDIUM else: return TaskComplexity.COMPLEX def select_pipeline( self, complexity: TaskComplexity ) -> list[str]: """根据复杂度选择执行路径的步骤列表。""" pipelines = { TaskComplexity.SIMPLE: [ "guardrail", "router", "executor", ], TaskComplexity.MEDIUM: [ "guardrail", "router", "memory", "executor", "auditor", ], TaskComplexity.COMPLEX: [ "guardrail", "router", "memory", "planner", "executor", "auditor", ], } return pipelines[complexity] def build_dag( self, query: str, pipeline_steps: list[str], ) -> list[DAGTask]: """根据执行路径和查询内容构建 DAG 任务列表。""" tasks = [] for i, step in enumerate(pipeline_steps): # 基本依赖：每步依赖前一步完成 depends_on = [tasks[i - 1].task_id] if i > 0 else [] # 特殊处理：executor 步骤可能有并行子任务 if step == "executor": # 检查是否需要并行工具调用 parallel_tools = self._extract_parallel_tools(query) if parallel_tools: # 创建并行子任务 for j, tool_name in enumerate(parallel_tools): sub_task = DAGTask( task_id=f"exec_{tool_name}", name=f"并行执行: {tool_name}", handler_name=tool_name, args={"query": query}, depends_on=[tasks[i - 1].task_id], # 依赖前一步 ) tasks.append(sub_task) continue # 跳过串行 executor 节点 task = DAGTask( task_id=f"step_{step}", name=f"步骤: {step}", handler_name=step, args={"query": query}, depends_on=depends_on, ) tasks.append(task) logger.info( "dag_built", query=query[:50], steps=len(tasks), pipeline=pipeline_steps, ) return tasks def _extract_parallel_tools(self, query: str) -> list[str]: """从查询中提取可并行执行的工具列表。""" # 简化版：检测并列关键词 parallel_keywords = { "天气": "weather_api", "餐厅": "restaurant_api", "新闻": "news_api", "酒店": "hotel_api", "地图": "map_api", "汇率": "exchange_api", } tools = [] for keyword, tool_name in parallel_keywords.items(): if keyword in query: tools.append(tool_name) return tools # ========== DAG 执行引擎 ========== class DAGExecutor: """DAG 并行编排执行引擎。 核心逻辑：按拓扑层级调度任务，同一层级内的任务并行执行。 """ def __init__(self, max_parallel: int = 5): self.max_parallel = max_parallel self._handlers: dict[str, Any] = {} self._semaphore = asyncio.Semaphore(max_parallel) def register_handler(self, name: str, handler): """注册任务处理函数。""" self._handlers[name] = handler async def execute_dag( self, tasks: list[DAGTask], context: dict[str, Any] | None = None, ) -> dict[str, Any]: """执行 DAG 任务列表。""" context = context or {} task_map = {t.task_id: t for t in tasks} results: dict[str, Any] = {} total_start = time.monotonic() # 计算拓扑层级（每层的任务互不依赖，可并行） layers = self._compute_topological_layers(tasks, task_map) logger.info( "dag_execution_start", total_tasks=len(tasks), layers=len(layers), ) # 按层执行 for layer_idx, layer_tasks in enumerate(layers): logger.info( "dag_layer_start", layer=layer_idx, tasks=len(layer_tasks), ) # 同一层的任务并行执行 layer_results = await asyncio.gather( *[self._execute_single_task(t, context, results) for t in layer_tasks], return_exceptions=True, ) # 收集结果 for task, result in zip(layer_tasks, layer_results): if isinstance(result, Exception): task.status = TaskStatus.FAILED task.error = str(result) results[task.task_id] = {"error": str(result)} logger.error( "task_failed", task_id=task.task_id, error=str(result), ) else: task.status = TaskStatus.SUCCESS task.result = result results[task.task_id] = result total_latency = (time.monotonic() - total_start) * 1000 success_count = sum( 1 for t in tasks if t.status == TaskStatus.SUCCESS ) logger.info( "dag_execution_complete", total_tasks=len(tasks), success=success_count, failed=len(tasks) - success_count, latency_ms=round(total_latency, 1), ) return { "results": results, "success_count": success_count, "failed_count": len(tasks) - success_count, "total_latency_ms": round(total_latency, 1), } async def _execute_single_task( self, task: DAGTask, context: dict[str, Any], completed_results: dict[str, Any], ) -> Any: """执行单个 DAG 任务，带限流和重试。""" handler = self._handlers.get(task.handler_name) if handler is None: # 没有注册的 handler，用默认模拟处理 async with self._semaphore: start = time.monotonic() await asyncio.sleep(0.05) # 模拟处理时间 task.latency_ms = round((time.monotonic() - start) * 1000, 1) return {"task": task.name, "status": "mock_success"} # 将前置任务的结果注入上下文 task_args = dict(task.args) for dep_id in task.depends_on: if dep_id in completed_results: task_args[f"dep_{dep_id}"] = completed_results[dep_id] # 带限流的执行 for attempt in range(task.max_retries + 1): try: async with self._semaphore: start = time.monotonic() result = await handler(task_args, context) task.latency_ms = round( (time.monotonic() - start) * 1000, 1 ) return result except Exception as e: task.retry_count = attempt + 1 logger.warning( "task_retry", task_id=task.task_id, attempt=attempt + 1, error=str(e), ) if attempt >= task.max_retries: raise def _compute_topological_layers( self, tasks: list[DAGTask], task_map: dict[str, DAGTask], ) -> list[list[DAGTask]]: """计算拓扑层级：将 DAG 分解为可并行执行的层。""" layers: list[list[DAGTask]] = [] completed_ids: set[str] = set() remaining = list(tasks) while remaining: # 找出所有依赖已满足的任务（即前置任务都已完成或无依赖） ready_tasks = [ t for t in remaining if all( dep in completed_ids or dep not in task_map for dep in t.depends_on ) ] if not ready_tasks: # 存在循环依赖或无法解析的依赖 logger.error("dag_deadlock", remaining=len(remaining)) # 将剩余任务标记为 SKIPPED for t in remaining: t.status = TaskStatus.SKIPPED t.error = "dependency_deadlock" break layers.append(ready_tasks) for t in ready_tasks: completed_ids.add(t.task_id) remaining = [t for t in remaining if t not in ready_tasks] return layers # ========== 段落级检索方案 ========== @dataclass class Passage: """段落级检索单元。""" passage_id: str content: str doc_id: str source: str position: int # 在文档中的位置（第几个段落） embedding: list[float] = field(default_factory=list) class PassageLevelSearcher: """段落级检索器：8 月的检索精度优化方案。 解决痛点：文档级检索丢失长文档中的关键段落。 """ def __init__(self, top_k_passages: int = 10, top_k_docs: int = 5): self.top_k_passages = top_k_passages self.top_k_docs = top_k_docs self._passages: list[Passage] = [] def split_document_to_passages( self, doc_id: str, content: str, source: str, min_passage_length: int = 100, ) -> list[Passage]: """将文档按段落拆分为检索单元。""" # 按自然段落分割 raw_paragraphs = content.split("\n\n") passages = [] position = 0 for para in raw_paragraphs: # 过滤过短的段落 if len(para.strip()) < min_passage_length: continue # 对超长段落再做 sentence 级拆分 if len(para) > 500: sentences = self._split_to_sentences(para) # 将连续的 sentence 合并为 200-400 token 的片段 chunks = self._merge_sentences_to_chunks( sentences, max_chunk_length=400 ) for chunk in chunks: p = Passage( passage_id=f"{doc_id}_p{position}", content=chunk, doc_id=doc_id, source=source, position=position, ) passages.append(p) position += 1 else: p = Passage( passage_id=f"{doc_id}_p{position}", content=para, doc_id=doc_id, source=source, position=position, ) passages.append(p) position += 1 logger.info( "document_split", doc_id=doc_id, total_passages=len(passages), original_length=len(content), ) return passages def _split_to_sentences(self, text: str) -> list[str]: """按句号、问号、感叹号分割句子。""" import re sentences = re.split(r'[。？！\.\?\!]\s*', text) return [s.strip() for s in sentences if len(s.strip()) > 20] def _merge_sentences_to_chunks( self, sentences: list[str], max_chunk_length: int = 400 ) -> list[str]: """将句子合并为固定长度的片段。""" chunks = [] current_chunk = "" for sentence in sentences: if len(current_chunk) + len(sentence) > max_chunk_length: if current_chunk: chunks.append(current_chunk.strip()) current_chunk = sentence else: current_chunk += sentence + "。" if current_chunk.strip(): chunks.append(current_chunk.strip()) return chunks async def search( self, query_embedding: list[float], max_results: int = 10, ) -> list[dict[str, Any]]: """段落级检索：返回最相关的段落，按文档聚合。""" # 实际项目中对接向量数据库的段落级搜索 # 这里用模拟数据演示 await asyncio.sleep(0.02) # 模拟检索延迟 # 模拟段落级检索结果 mock_results = [ { "passage_id": "doc_1_p3", "content": "关于 asyncio 的最佳实践段落...", "doc_id": "doc_1", "source": "tech_blog", "score": 0.92, "position": 3, }, { "passage_id": "doc_2_p1", "content": "asyncio 事件循环的核心机制...", "doc_id": "doc_2", "source": "official_doc", "score": 0.88, "position": 1, }, ] # 按文档聚合：同一个文档的多个段落合并展示 aggregated = self._aggregate_by_document(mock_results) logger.info( "passage_search_complete", query_dim=len(query_embedding), passages=len(mock_results), documents=len(aggregated), ) return aggregated[:max_results] def _aggregate_by_document( self, passage_results: list[dict] ) -> list[dict[str, Any]]: """将段落结果按文档聚合，保留最佳段落作为代表。""" doc_groups: dict[str, list[dict]] = {} for p in passage_results: doc_id = p["doc_id"] if doc_id not in doc_groups: doc_groups[doc_id] = [] doc_groups[doc_id].append(p) aggregated = [] for doc_id, passages in doc_groups.items(): # 取该文档中 score 最高的段落 best_passage = max(passages, key=lambda p: p["score"]) aggregated.append({ "doc_id": doc_id, "source": best_passage.get("source", ""), "best_passage": best_passage["content"], "best_score": best_passage["score"], "total_matching_passages": len(passages), "all_passages": passages, }) # 按最高段落 score 降序排序 aggregated.sort(key=lambda d: d["best_score"], reverse=True) return aggregated # ========== 月度复盘工具 ========== @dataclass class MonthlyGoal: """月度目标追踪。""" name: str category: str # agent, rag, infra, learning status: str # done, partial, not_started completion_pct: float # 0-100 lessons: str # 收获/教训 next_action: str # 8 月后续行动 class MonthlyReviewPlanner: """月度复盘与 8 月规划工具。""" def __init__(self): self.july_goals: list[MonthlyGoal] = [] self.august_goals: list[MonthlyGoal] = [] self._load_july_review() def _load_july_review(self): """加载 7 月目标复盘。""" self.july_goals = [ MonthlyGoal( name="Agent 分层架构", category="agent", status="done", completion_pct=100, lessons="分层解耦+审计节点是 Agent 生产化的基础", next_action="升级为 DAG 并行编排", ), MonthlyGoal( name="RAG 延迟压缩", category="rag", status="done", completion_pct=100, lessons="逐段定位瓶颈，逐轮定向爆破", next_action="段落级检索提升精度", ), MonthlyGoal( name="混合检索落地", category="rag", status="done", completion_pct=90, lessons="稀疏+稠密双编码，精度提升 17%", next_action="自适应权重调参 + DiskANN 评估", ), MonthlyGoal( name="Agent 并行编排", category="agent", status="partial", completion_pct=30, lessons="串行架构的并行化改造比预期复杂", next_action="实现 DAG 执行引擎", ), MonthlyGoal( name="动态路由引擎", category="agent", status="not_started", completion_pct=0, lessons="固定流水线对简单查询浪费步骤", next_action="实现复杂度评估+自适应路径", ), MonthlyGoal( name="async 工具箱", category="infra", status="done", completion_pct=100, lessons="限流+Task管理+HTTP单例是基础能力", next_action="增加熔断器模式", ), ] self.august_goals = [ MonthlyGoal( name="DAG 并行编排引擎", category="agent", status="not_started", completion_pct=0, lessons="", next_action="实现拓扑排序调度 + 层级并行执行", ), MonthlyGoal( name="动态路由引擎", category="agent", status="not_started", completion_pct=0, lessons="", next_action="实现复杂度评估器 + 三级路径选择", ), MonthlyGoal( name="段落级检索", category="rag", status="not_started", completion_pct=0, lessons="", next_action="文档拆分为 passage + 独立向量化", ), MonthlyGoal( name="Agent 记忆持久化", category="agent", status="not_started", completion_pct=0, lessons="", next_action="跨 session 的长期记忆存储和检索", ), MonthlyGoal( name="Scalar 量化落地", category="rag", status="not_started", completion_pct=0, lessons="", next_action="Qdrant Scalar Quantization 部署和验证", ), MonthlyGoal( name="全链路可观测性", category="infra", status="not_started", completion_pct=0, lessons="", next_action="Agent 全链路 trace + Token 成本追踪", ), ] def get_review_summary(self) -> dict[str, Any]: """7 月复盘摘要。""" done = [g for g in self.july_goals if g.status == "done"] partial = [g for g in self.july_goals if g.status == "partial"] not_done = [g for g in self.july_goals if g.status == "not_started"] avg_completion = sum(g.completion_pct for g in self.july_goals) / len(self.july_goals) return { "july_total_goals": len(self.july_goals), "completed": len(done), "partial": len(partial), "not_started": len(not_done), "avg_completion_pct": round(avg_completion, 1), "top_lessons": [ g.lessons for g in done if g.lessons ], "carry_over": [ g.name for g in partial + not_done ], } def get_august_plan(self) -> dict[str, Any]: """8 月规划摘要。""" priorities = { "P0": [g for g in self.august_goals if g.category in ["agent", "rag"]], "P1": [g for g in self.august_goals if g.category == "infra"], } return { "august_total_goals": len(self.august_goals), "p0_count": len(priorities["P0"]), "p1_count": len(priorities["P1"]), "p0_goals": [g.name for g in priorities["P0"]], "p1_goals": [g.name for g in priorities["P1"]], "focus_areas": list(set(g.category for g in self.august_goals)), } async def main(): """演示：动态路由 + DAG 执行 + 月度复盘。""" # 1. 动态路由 router = DynamicRouter() queries = [ "北京今天天气怎么样", # simple "帮我对比 Redis 和 Milvus 的性能差异", # medium "规划下周出差行程，查天气、订酒店、查交通", # complex ] for query in queries: complexity = router.evaluate_complexity(query) pipeline = router.select_pipeline(complexity) dag_tasks = router.build_dag(query, pipeline) logger.info( "dynamic_routing", query=query[:50], complexity=complexity.value, pipeline=pipeline, dag_tasks=len(dag_tasks), ) # 2. DAG 执行 executor = DAGExecutor(max_parallel=5) # 注册模拟 handler async def mock_handler(args: dict, ctx: dict) -> dict: await asyncio.sleep(0.05) return {"result": f"mock_{args.get('query', '')[:20]}"} for step in ["guardrail", "router", "executor", "memory", "planner", "auditor"]: executor.register_handler(step, mock_handler) # 构建并执行 DAG tasks = router.build_dag(queries[2], router.select_pipeline(TaskComplexity.COMPLEX)) result = await executor.execute_dag(tasks) logger.info("dag_result", **result) # 3. 月度复盘 planner = MonthlyReviewPlanner() review = planner.get_review_summary() august_plan = planner.get_august_plan() logger.info("july_review", **review) logger.info("august_plan", **august_plan) if __name__ == "__main__": asyncio.run(main())

四、边界分析与架构权衡

DAG 编排 vs 固定流水线：DAG 编排灵活度高，但调试复杂度也高。固定流水线串行执行，行为可预测、日志易追踪；DAG 编排并行执行，任务间依赖关系动态变化，一个节点失败可能触发多条路径的异常处理。我的策略是：第一版 DAG 只支持两层并行（预计算层 + 执行层），不搞复杂的 DAG 嵌套。等两层并行稳定了，再增加层级。

动态路由的误判风险：复杂度评估器是规则驱动的，不是模型驱动的。规则简单可靠，但覆盖不了所有场景。"帮我查一下 Python 的装饰器用法"——规则判断为 SIMPLE（单工具），但实际上用户可能期待详细的代码示例和对比分析，需要 MEDIUM 级别的路径。更精细的方案是用小模型做意图分类，但增加了延迟和成本。

段落级检索的向量膨胀：一篇 100 段的文档变成 100 个向量，10 万篇文档就是 1000 万个向量。向量数量增加 10 倍，索引构建时间、内存占用、检索延迟都受影响。解决方案是：段落级索引只对高频查询的热门文档启用，冷门文档保留文档级索引。这种"分级索引"策略在向量数量和检索精度之间做了折中。

8 月目标数量控制：规划了 6 个目标，但 8 月只有 4 周。我的原则是 P0 目标最多 3 个（DAG 编排、动态路由、段落级检索），P1 目标按精力弹性推进（记忆持久化、量化落地、可观测性）。宁可少做几个但做完，不要贪多但每个都半成品。

五、总结

7 月收官，10 篇文章写完。回头看，这个月在 Agent 和 RAG 上走了很远的路，但还有更长的路要走。

7 月的硬产出是六件事：Agent 分层架构、RAG 延迟压缩、混合检索、Prompt 量化方法论、Redis 向量运维、async 工具箱。每一件都是从痛点出发、有代码落地、有数据验证的实打实的成果。没有"值得关注式发现"，没有"推荐阅读级突破"，只有一个个具体的问题被一个个具体的方案解决。

7 月的遗憾也是三个：Agent 并行编排只实现了 30%、动态路由引擎没启动、长文档检索精度仍有瓶颈。这些遗憾不是失败，是"未完成"——它们已经从痛点变成了明确的 8 月规划。

8 月的方向很清晰：DAG 并行编排是 Agent 进化的下一步，动态路由引擎是效率优化的关键，段落级检索是精度跃升的路径。三个方向都是从 7 月的遗憾中生长出来的——每一个未解决的问题，都指向了下一步的明确方向。

做技术规划的本质不是列清单，是排优先级。6 个目标看起来很多，但 P0 只有 3 个。3 个 P0 目标在 4 周内做完是有把握的——每个目标 1 周的核心开发 + 1 周的验证打磨。P1 目标看精力弹性推进，做不完就顺延到 9 月。不焦虑，不贪多，一步步走。

7 月，充实收官。8 月，继续迭代。

资料说明

本文中的协议、版本、性能、成本和行业趋势应以可核验的一手资料为准。未标注统计口径的比例、时间表和预测仅作工程讨论，不应视为行业事实。可参考 0731 资料来源索引，并在发布前将具体来源贴到对应断言之后。