优化流程之前，需要先定义质量要求、成本口径和延迟口径。缓存、模型分流和失败回退都可能改善某项指标，也可能带来新的问题。以知识问答为例，系统先检查可复用结果，再根据复杂度与预算选择执行路径，并记录每次运行的状态和消耗。
目标、状态、执行边界和验收标准
| 项目 | 定义 |
|---|---|
| 目标 | 在逻辑预算内返回答复，记录成功状态、调用次数、路径和延迟。 |
| 状态 | budget限定可用单位；units记录已消耗单位；calls记录调用数；route记录缓存或模型路径；fallback记录回退；latency_ms记录耗时。 |
| 执行边界 | fast消耗一个单位，deep消耗三个单位；deep只在预算至少为四时使用，以预留回退；失败调用也消耗单位；缓存按作用范围和资料版本隔离。 |
| 验收标准 | 任何运行消耗不超预算；命中缓存不调用模型；可回退时完成；无预算时停止；不同作用范围不会复用同一缓存。 |
方法与设计思想
评估：明确分母与样本范围
成功率等于完成的任务数除以总任务数，预算耗尽也计入总数。延迟使用perf_counter记录，并用nearest-rank计算P95。本脚本只有五条固定教学样本，结果仅用于验证统计方法。
质量、过程与运行表现要分别检查。完成状态不代表答案语义正确，低延迟也可能来自缓存命中或提前失败。生产统计应按路径和成功状态分组，并注明样本数与时间窗口。
资源感知路由：先保证预算，再选择能力
示例以是否包含“比较”判断复杂度，预算允许时选择deep，否则选择fast。这是透明的教学规则，需要用真实任务验证误分流情况。
单位是自定义调用预算，不是货币或真实token数量。失败调用也占用单位，避免回退后账面看起来没有成本。接入服务时，应使用供应商的实际用量及结算口径，并考虑预留与结算差异。
缓存：复用相同语义条件下的结果
缓存键包含作用范围、资料版本、问题和策略版本，避免不同用户或资料版本混用。命中缓存可以在零新增调用预算下返回结果。
缓存仍需要有效期、容量、失效规则和权限检查。模型升级或任务标准变化时，缓存策略也需要同步变化。
回退：明确失败类型和剩余预算
首选路径出现TimeoutError后，只在剩余单位足够时回退到fast。预算不足则终止，不继续调用。真实服务的异常类型需要按具体模型适配，认证错误和输入错误通常不应自动回退。
为什么加入这些机制
| 机制 | 解决的问题 | 验证方式 |
|---|---|---|
| 评估记录 | 让成本和延迟有可检查的口径 | 检查五条结果的分母、调用数和P95算法 |
| 资源路由 | 控制不同任务的投入 | 低预算走fast，复杂且预算足够时走deep |
| 缓存 | 减少重复计算 | 第二次相同请求的调用数为零；其他用户重新调用 |
| 预算回退 | 在约束内恢复首选路径失败 | deep失败后总消耗为四；一单位失败后停止 |
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
将下面完整代码保存为flow06.py，运行python flow06.py。默认模式不需要模型凭据，使用固定数据验证流程和异常分支。
import hashlib
import json
import math
import os
import time
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
EVIDENCE = "RAG使用外部证据；持久记忆保存跨任务信息。"
CACHE = {}
def make_chain(fail=False, tier="fast"):
prompt = ChatPromptTemplate.from_template(
"根据资料回答问题，不增加无依据结论。资料：{evidence}；问题：{question}"
)
if fail:
def unavailable(_):
raise TimeoutError("教学用：首选模型暂时不可用")
return RunnableLambda(unavailable)
if os.getenv("MODEL_ID_FAST"):
variable = "MODEL_ID_DEEP" if tier == "deep" else "MODEL_ID_FAST"
model = init_chat_model(os.environ[variable])
else:
model = FakeListChatModel(responses=[EVIDENCE])
return prompt | model | StrOutputParser()
def run(question, budget, fail_primary=False, scope="demo", corpus_version="v1"):
started = time.perf_counter()
# 预算单位是自定义调用单位，不是货币或真实token数量。
key = hashlib.sha256(json.dumps([scope, corpus_version, question,
"policy-v1"], ensure_ascii=False).encode()).hexdigest()
if key in CACHE:
return {"answer": CACHE[key], "status": "completed", "units": 0,
"calls": 0, "route": "cache", "fallback": False,
"latency_ms": (time.perf_counter() - started) * 1000}
if budget < 1:
return {"answer": "", "status": "budget_exhausted", "units": 0,
"calls": 0, "route": "none", "fallback": False,
"latency_ms": (time.perf_counter() - started) * 1000}
complex_task = "比较" in question
primary_cost = 3 if complex_task and budget >= 4 else 1
route = "deep" if primary_cost == 3 else "fast"
units, calls, fallback = primary_cost, 1, False
payload = {"question": question, "evidence": EVIDENCE}
try:
answer = make_chain(fail_primary, tier=route).invoke(payload)
except TimeoutError:
if budget - units < 1:
return {"answer": "", "status": "budget_exhausted", "units": units,
"calls": calls, "route": route, "fallback": False,
"latency_ms": (time.perf_counter() - started) * 1000}
units += 1
calls += 1
fallback = True
answer = make_chain().invoke(payload)
CACHE[key] = answer
return {"answer": answer, "status": "completed", "units": units,
"calls": calls, "route": route, "fallback": fallback,
"latency_ms": (time.perf_counter() - started) * 1000}
def evaluate(rows):
values = sorted(row["latency_ms"] for row in rows)
return {"n": len(rows), "success_rate": sum(
r["status"] == "completed" for r in rows) / len(rows),
"p95_ms": values[math.ceil(0.95 * len(values)) - 1],
"total_units": sum(r["units"] for r in rows),
"total_calls": sum(r["calls"] for r in rows),
"definition": "nearest-rank；固定教学样本；不代表生产表现"}
if __name__ == "__main__":
rows = [run("RAG是什么？", 1), run("RAG是什么？", 0),
run("比较RAG和记忆", 4, fail_primary=True),
run("新问题", 1, fail_primary=True), run("零预算新问题", 0)]
assert rows[0]["units"] == 1
assert rows[1]["route"] == "cache" and rows[1]["calls"] == 0
assert rows[2]["fallback"] and rows[2]["units"] == 4
assert rows[3]["status"] == "budget_exhausted" and rows[3]["calls"] == 1
assert rows[4]["calls"] == 0
assert all(r["units"] <= b for r, b in zip(rows, [1, 0, 4, 1, 0]))
assert run("RAG是什么？", 1, scope="another-user")["calls"] == 1
print(json.dumps({"metrics": evaluate(rows), "rows": rows}, ensure_ascii=False, indent=2))
run先检查缓存，再检查预算并选择路径。调用前确定所需单位，失败时判断是否能承担回退成本。evaluate根据全部运行记录统计成功率、总消耗和延迟分位数。真实模型可通过MODEL_ID_FAST和MODEL_ID_DEEP接入，但模拟超时分支仍是人为构造的测试条件。
接入两档真实模型
安装所选供应商集成包并配置认证环境变量后，分别设置MODEL_ID_FAST和MODEL_ID_DEEP为供应商:模型名称。两个变量应同时配置。逻辑预算仍按一单位和三单位计算，真实费用需要另行采集，不能混用这两个口径。
验收与验证结果
五条样本覆盖首次调用、缓存命中、deep失败后回退、fast失败且没有回退预算、零预算新任务。另一个作用范围验证缓存隔离。默认两档路径均使用固定测试模型，因此无法据此证明质量差异或模型速度差异。
Google ADK实现与LangChain对照
成本、缓存和评估算法属于业务规则，不需要为两个框架分别复制一套。ADK示例复用本篇flow06.py中的run与evaluate，再通过Step把结果提交到会话状态。保存两个文件到同一目录后运行adk06.py。
| 比较项 | LangChain／LangGraph | Google ADK |
|---|---|---|
| 业务算法 | run实现预算、缓存与回退 | 复用相同run，确保规则一致 |
| 结果记录 | 返回普通字典用于评估 | EventActions.state_delta保存record与status |
| 统计口径 | 主示例五个固定样本 | 对照示例四个固定样本；不能直接比较成功率 |
| 模型用量 | 需采集真实模型usage与费用 | 也需采集真实模型用量，框架选择不改变成本定义 |
Google ADK完整代码
对照代码已在google-adk==2.11.0与Python 3.12环境运行。安装python -m pip install google-adk==2.11.0，保存为adk06.py后运行。默认不调用真实模型。本篇需把上面的flow06.py与adk06.py保存在同一目录，同时安装LangChain依赖。
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)
from flow06 import run, evaluate
def answer(state):
record = run(state["question"], state["budget"],
fail_primary=state.get("fail_primary", False), scope=state["scope"])
return {"record": record, "status": record["status"]}
async def main():
rows = []
samples = [("RAG是什么", 1, False), ("RAG是什么", 0, False),
("比较RAG和记忆", 4, True), ("新问题", 1, True)]
for index, (question, budget, fail) in enumerate(samples):
result = await run_state(Step(name="resource", action=answer), {
"question": question, "budget": budget, "fail_primary": fail, "scope": "adk-demo",
}, session_id=f"s{index}")
rows.append(result["record"])
assert rows[1]["route"] == "cache" and rows[1]["calls"] == 0
assert rows[2]["fallback"] and rows[2]["units"] == 4
assert rows[3]["status"] == "budget_exhausted"
print(json.dumps(evaluate(rows), ensure_ascii=False, indent=2))
if __name__ == "__main__":
asyncio.run(main())
对照检查缓存零调用、回退消耗四单位和预算不足停止。没有做框架性能排名，因为样本和测量目标并不支持该结论。
接口与版本边界
对照采用SequentialAgent、ParallelAgent、LoopAgent或BaseAgent模板接口，以便比较相同控制思想。ADK 2.x官方文档已引入图工作流和动态工作流，新项目应评估对应机制；这里的版本运行通过不代表模板API是所有新项目的首选。两种框架的运行结果不能直接等同于真实模型效果。
扩展边界
未测量真实token、费用和模型SLA；没有生产质量评审、缓存持久化或实际限流测试。接入真实模型后，需要记录供应商用量，并用同一标注任务集比较路由前后的质量、成本和延迟。