复杂任务的第一步，是确定交付结果和执行约束。只有这些条件明确以后，才有依据选择顺序、分支、并行或动态规划。以“比较RAG与记忆”为例，系统需要选择必要的资料类别，取得结果并生成可检查的对比，而不是不断增加调用次数。
目标、状态、执行边界和验收标准
| 项目 | 定义 |
|---|---|
| 目标 | 针对问题选择必要的资料任务，生成带任务ID的结果汇总。 |
| 状态 | goal保存需求；plan保存任务ID、优先级和依赖；results保存各分支成果；status区分待输入与完成。 |
| 执行边界 | 仅允许rag与memory两类任务，最多两个；本执行器只接受独立任务；空需求不触发模型；未知任务或依赖直接拒绝。 |
| 验收标准 | 有效任务全部返回；任务ID无重复；输出顺序符合优先级；非法任务被拒绝；空需求进入needs_input。 |
方法与设计思想
提示链：逐步建立可检查的结果
提示链把复杂任务拆成有明确输入输出的步骤。这里采用“形成计划 → 解析结构 → 校验边界 → 执行 → 汇总”的顺序。计划的原始文本不能直接决定执行，必须先经过结构与业务规则检查。
思想是让每一步承担单一责任，并在进入下一步之前确认结果可用。模型输出错误时，错误应停在计划阶段。
规划与路由：先提出候选，再确定能执行的动作
规划回答“需要做哪些任务”，路由回答“这些任务由哪些组件处理”。模型可以提出rag或memory任务，但可用任务范围由程序决定。Plan与Task模型限制结构和数量，validate_plan限制依赖和重复ID。
示例拒绝带依赖的计划，因为后面的并行执行器没有拓扑调度能力。如果以后需要支持依赖，应先实现依赖校验和分层调度，再放开输入规则。
并行化与优先级：处理独立部分，控制结果组织
RunnableParallel让独立分支同时执行。汇总阶段等所有结果返回，再按优先级组织内容。优先级在这里决定展示顺序，不保证并行任务的启动先后；若需要严格调度，必须加入队列或分批执行机制。
并行减少的是独立任务的等待时间，不减少任务本身的数量。收益需要在真实延迟和限流条件下测量。
为什么加入这些机制
| 机制 | 解决的问题 | 验证方式 |
|---|---|---|
| 提示链 | 阻止未校验计划直接进入执行 | 提交非法JSON，确认执行分支不会运行 |
| 规划与路由 | 把自然语言需求映射为有限任务集合 | 分别提出单类、双类、未知类需求并检查任务选择 |
| 并行化 | 允许独立检索同时推进 | 为分支加入可控延迟，比较并行与串行耗时 |
| 优先级 | 保持结果顺序可预测 | 打乱计划顺序，检查汇总仍按priority排序 |
LangChain与LangGraph代码实现
主代码在Python 3.12.14、LangChain 1.4.3、langchain-core 1.6.6、LangGraph 1.2.12、Pydantic 2.13.5下运行验证。建议使用独立虚拟环境：
python3.12 -m venv .venv
source .venv/bin/activate
python -m pip install langchain==1.4.3 langchain-core==1.6.6 langgraph==1.2.12 pydantic==2.13.5
将下面完整代码保存为flow01.py，运行python flow01.py。默认模式不需要模型凭据，使用固定数据验证流程和异常分支。
import json
import os
from typing import Literal
from pydantic import BaseModel, Field
from langchain.chat_models import init_chat_model
from langchain_core.language_models.fake_chat_models import FakeListChatModel
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableLambda, RunnableParallel
DOCS = {
"rag": "RAG先检索证据，再组织上下文生成答案。",
"memory": "记忆保存交互经验；跨会话保存需要外部存储。",
}
class Task(BaseModel):
id: Literal["rag", "memory"]
priority: int = Field(ge=1, le=5)
depends_on: list[str] = Field(default_factory=list)
class Plan(BaseModel):
tasks: list[Task] = Field(min_length=1, max_length=2)
def validate_plan(text: str) -> Plan:
plan = Plan.model_validate_json(text)
ids = [t.id for t in plan.tasks]
if len(ids) != len(set(ids)):
raise ValueError("任务ID重复")
for task in plan.tasks:
if task.depends_on:
raise ValueError("本执行器仅支持独立任务")
return plan
def build_planner():
prompt = ChatPromptTemplate.from_messages([
("system", '只输出JSON，结构为{{"tasks":[{{"id":"rag",'
'"priority":1,"depends_on":[]}}]}}。任务id仅允许rag或memory，'
'根据问题选择必要任务；priority为1到5，数字越小越先处理。'),
("human", "{question}"),
])
model = (init_chat_model(os.environ["MODEL_ID"]) if os.getenv("MODEL_ID")
else FakeListChatModel(responses=[json.dumps({"tasks": [
{"id": "memory", "priority": 2},
{"id": "rag", "priority": 1},
]})]))
return prompt | model | StrOutputParser() | RunnableLambda(validate_plan)
def run(question: str) -> dict:
state = {"goal": question.strip(), "status": "pending"}
if not state["goal"]:
return {**state, "status": "needs_input", "results": {}}
plan = build_planner().invoke({"question": state["goal"]})
ordered = sorted(plan.tasks, key=lambda t: t.priority)
workers = {
t.id: RunnableLambda(lambda _, key=t.id: DOCS[key]) for t in ordered
}
# 优先级确定展示顺序；独立分支同时执行，不保证启动先后。
results = RunnableParallel(workers).invoke({})
answer = "\n".join(f"{t.id}: {results[t.id]}" for t in ordered)
return {**state, "plan": plan.model_dump(), "results": results,
"answer": answer, "status": "completed"}
if __name__ == "__main__":
result = run("比较RAG与记忆")
assert result["status"] == "completed"
if not os.getenv("MODEL_ID"):
assert set(result["results"]) == {"rag", "memory"}
assert result["answer"].startswith("rag:")
assert run(" ")["status"] == "needs_input"
for invalid in ['{"tasks":[]}',
'{"tasks":[{"id":"shell","priority":1}]}',
'{"tasks":[{"id":"rag","priority":1,"depends_on":["memory"]}]}']:
try:
validate_plan(invalid)
except ValueError:
continue
raise AssertionError("无效计划未被拒绝")
print(json.dumps(result, ensure_ascii=False, indent=2))
build_planner构造提示、模型、文本解析和业务校验链。run先处理空需求，再执行计划；每个任务ID对应一个Runnable，只有允许集合中的任务才能进入并行组件。最终状态保存计划、各分支结果和汇总，便于定位问题。
接入真实模型
代码通过init_chat_model读取MODEL_ID。安装所选供应商的集成包，配置其认证环境变量，再把MODEL_ID设为供应商:模型名称即可切换。供应商和模型需支持相应功能；第二篇需要工具调用能力。敏感参数放在环境变量中，不能写入正文或源码。真实模型输出允许变化，结构与边界检查保持相同。
验收与验证结果
默认固定计划包含rag与memory，断言检查两类结果完整、rag排在前面。三个异常输入分别覆盖空任务列表、未知任务和不支持的依赖。固定模型只验证编排链路；真实规划质量需要额外使用有标注的需求集检查漏任务、冗余任务和错误分流。
Google ADK实现与LangChain对照
Google ADK版本使用SequentialAgent连接计划、并行任务和汇总，ParallelAgent执行两个独立分支。候选计划从candidate_tasks输入提供，规则检查与LangChain版本的执行边界对应；动态模型规划仍需另加LlmAgent与结构校验。
| 比较项 | LangChain／LangGraph | Google ADK |
|---|---|---|
| 顺序控制 | 提示链与Runnable组合 | SequentialAgent按sub_agents顺序推进 |
| 并行控制 | RunnableParallel返回命名结果 | ParallelAgent分支写入各自的result字段 |
| 计划来源 | 固定模型或真实模型产出结构化计划 | 示例使用外部提供的候选计划；可另加模型生成阶段 |
| 业务边界 | Pydantic与validate_plan检查 | plan函数检查数量、ID、依赖和优先级 |
Google ADK完整代码
对照代码已在google-adk==2.11.0与Python 3.12环境运行。安装python -m pip install google-adk==2.11.0，保存为adk01.py后运行。默认不调用真实模型。
import asyncio
import inspect
import json
from collections.abc import Callable
from typing import Any
from google.adk.agents import BaseAgent, SequentialAgent, ParallelAgent, LoopAgent
from google.adk.events import Event, EventActions
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
class Step(BaseAgent):
action: Callable[..., Any]
async def _run_async_impl(self, ctx):
delta = self.action(ctx.session.state)
if inspect.isawaitable(delta):
delta = await delta
yield Event(author=self.name, actions=EventActions(state_delta=delta))
async def run_state(agent, state, session_id="s1", user_id="u1", service=None):
service = service or InMemorySessionService()
await service.create_session(app_name="demo", user_id=user_id,
session_id=session_id, state=state)
runner = Runner(agent=agent, app_name="demo", session_service=service)
async for _ in runner.run_async(user_id=user_id, session_id=session_id,
new_message=types.Content(role="user", parts=[types.Part(text="执行任务")])):
pass
session = await service.get_session(app_name="demo", user_id=user_id,
session_id=session_id)
return dict(session.state)
DOCS = {"rag": "RAG先检索再生成。", "memory": "记忆保存跨任务经验。"}
def plan(state):
if not state["goal"].strip():
return {"status": "needs_input", "tasks": []}
tasks = state["candidate_tasks"]
ids = [task["id"] for task in tasks]
if not 1 <= len(tasks) <= 2 or len(ids) != len(set(ids)):
raise ValueError("任务数量或ID无效")
for task in tasks:
if task["id"] not in DOCS or task.get("depends_on"):
raise ValueError("未知任务或不支持的依赖")
if type(task["priority"]) is not int or not 1 <= task["priority"] <= 5:
raise ValueError("优先级无效")
return {"tasks": sorted(tasks, key=lambda t: t["priority"]), "status": "running"}
def worker(key):
def action(state):
if any(task["id"] == key for task in state["tasks"]):
return {f"result_{key}": DOCS[key]}
return {}
return Step(name=f"worker_{key}", action=action)
def merge(state):
if state["status"] == "needs_input":
return {}
answer = "\n".join(f"{task['id']}: {state['result_' + task['id']]}"
for task in state["tasks"])
return {"answer": answer, "status": "completed"}
def build():
return SequentialAgent(name="pipeline", sub_agents=[
Step(name="plan", action=plan),
ParallelAgent(name="parallel", sub_agents=[worker("rag"), worker("memory")]),
Step(name="merge", action=merge),
])
async def main():
tasks = [{"id": "memory", "priority": 2}, {"id": "rag", "priority": 1}]
result = await run_state(build(), {"goal": "比较RAG与记忆", "candidate_tasks": tasks})
assert result["status"] == "completed" and result["answer"].startswith("rag:")
empty = await run_state(build(), {"goal": "", "candidate_tasks": []})
assert empty["status"] == "needs_input"
try:
await run_state(build(), {"goal": "非法任务", "candidate_tasks": [{"id": "shell", "priority": 1}]})
except ValueError:
pass
else:
raise AssertionError("未知任务未被拒绝")
print(json.dumps(result, ensure_ascii=False, indent=2))
if __name__ == "__main__":
asyncio.run(main())
ADK测试覆盖正常汇总、空需求和未知任务。两种实现都按优先级组织结果，但没有把优先级当作并行启动顺序。模型规划与执行器验证需要分别统计。
接口与版本边界
对照采用SequentialAgent、ParallelAgent、LoopAgent或BaseAgent模板接口，以便比较相同控制思想。ADK 2.x官方文档已引入图工作流和动态工作流，新项目应评估对应机制；这里的版本运行通过不代表模板API是所有新项目的首选。两种框架的运行结果不能直接等同于真实模型效果。
扩展边界
这是有约束的工作流执行器。它没有任意工具调用、自动扩展任务、依赖图调度或持久任务队列。真实模型模式会发送问题给配置的模型服务，运行前应明确资料权限。