L20 生产部署（下）：Foundry 托管 + 模型路由 + 评估门

• 难度 ⭐⭐⭐⭐⭐（高级）
• 预计阅读 35 分钟
• 代码量 含 Foundry hosted agent + 模型路由 + 评估门
• 先修 L16（MAF）+ L19（生产部署上）

课程目标：理解 Foundry 的部署模式、可观测性、成本优化、HITL、生产安全 MCP 集成等关键实践，掌握模型路由 + 缓存 + 评估门的工业级部署三件套。

一、Demo 和生产隔着一道银河#

Deploying Scalable Agents 开篇一句话震撼：

"It is not the right way to run an agent that thousands of customers depend on at 3 a.m."

你在本地跑通了一个 Agent：

agent.run("帮我订明天上午的会议室")
# 输出：好的，已帮你预订明天 9-10 点的会议室 A

Demo 成功。但部署到生产后：

• 怎么知道每天有多少人用？

• 怎么知道平均花费多少 token？

• 怎么知道 Agent 哪一步出错最多？

• Agent 出现循环调用怎么办？

• 新版本发布怎么知道没回归？

• 服务挂了怎么排查？

L19（AI Agents in Production）已经讲过可观测性。但 L20 关注的是"上线部署"——怎么把 Agent 变成一个 24/7 在线、可扩展、有护网、有版本控制的生产服务。

二、用一个比喻：Agent 部署 = 开连锁餐厅#

把 Agent 比作连锁餐厅：

没有运营手册的餐厅 = 没有可观测性的生产 Agent——你不知道发生了什么，只能事后救火。

三、Prototype vs Production：Deploying Scalable Agents 的关键对比#

Deploying Scalable Agents 一针见血：

生产 = 模型周围那 80% 的工作——hosting、identity、state、failure、cost、quality、trust。

四、三大部署模式#

Deploying Scalable Agents 列出三大部署模式：

4.1 Client-Hosted（客户端托管）#

Agent 对象跑在你的应用进程里，你的代码直接调模型。

# 你自己的后端服务
from agent_framework.azure import AzureOpenAIChatClient
from azure.identity import DefaultAzureCredential

client = AzureOpenAIChatClient(credential=DefaultAzureCredential())
agent = client.create_agent(
    instructions="You are a customer support agent",
    tools=[search_order, create_ticket]
)

# 你的 API 端点
async def support_endpoint(user_query: str):
    return await agent.run(user_query)

适用：需要完全控制循环、要嵌入现有后端。

代价：你自己管 scaling、state、resilience。

4.2 Hosted Agents（Foundry 托管）#

Agent 作为资源注册到 Foundry——Foundry 帮你管理 reasoning loop、threads、content safety、RBAC。

# 用 Foundry Agent Service SDK 注册 Agent
from azure.ai.projects import AIProjectClient
from azure.identity import DefaultAzureCredential

project_client = AIProjectClient.from_connection_string(
    credential=DefaultAzureCredential(),
    conn_str="..."
)

agent = project_client.agents.create_agent(
    model="gpt-5-mini",
    name="ContosoSupportAgent",
    instructions="You are customer support. Use the available tools.",
    tools=[search_order, create_ticket]
)

适用：想要耐久性、内置可观测性、治理。

代价：低层控制权少。

4.3 Agent Workflows（工作流编排）#

多个 Agent 组合成图——顺序步骤、分支、HITL 审批、checkpoint：

workflow = WorkflowBuilder()
workflow.add_edge(intake_agent, lookup_agent)
workflow.add_edge(lookup_agent, response_agent, condition=needs_human_review)
workflow.add_edge(response_agent, end, condition=approved)

适用：跨多个专业 Agent、需要中途审批。

五、Agent 生命周期#

Deploying Scalable Agents 给出了完整的 Agent 生命周期：

创建/开发 → 版本化 → 离线评估 → 部署 → 在线观测
   ↑                                          ↓
   └──────── 不通过评估门 ──────────── 退役旧版本
                                    ↑
                                在线观测反馈到测试集

关键原则：

"Offline evaluation is a gate, not an afterthought."

新版本 Agent 不通过评估门就不上线。

六、四大扩展手段#

6.1 无状态请求处理#

不在进程内存里保留 per-user state；把 conversation threads 持久化到 Foundry thread store。

# 每次请求都从 store 恢复 thread
async def handle_request(thread_id: str, user_query: str):
    thread = await threadstore.get(thread_id)
    response = await agent.run(user_query, thread=thread)
    await threadstore.save(thread_id, thread)
    return response

好处：任意实例都能处理任意请求——水平扩展的基础。

6.2 模型路由#

简单请求走小模型，复杂请求才走大模型：

User query
   ↓
Cache hit? ── yes ──> 返回缓存
   ↓ no
Complexity? ── simple ──> Small model (gpt-5-nano)
              ── complex ──> Large model (gpt-5-mini)
   ↓
Response → 缓存 + trace

async def handle_support_request(query: str, customer_id: str):
    cached = response_cache.get(normalize(query))
    if cached:
        return cached

    model = "gpt-5-nano" if is_simple(query) else "gpt-5-mini"

    with tracer.start_as_current_span("support_request") as span:
        span.set_attribute("routed.model", model)
        response = await support_agent.run(query, model=model)

    response_cache.set(normalize(query), response.text)
    return response.text

6.3 响应缓存#

缓存相似查询的响应——"怎么重置密码？" 这种 FAQ 一辈子都有人问。

6.4 并发与背压#

限制并发、用重试 + 指数退避、优雅失败：

async with semaphore(limit=100):  # 最多 100 个并发
    try:
        return await agent.run(query)
    except RateLimitError:
        await asyncio.sleep(2 ** retry_count)  # 指数退避
        retry_count += 1

七、可观测性：Deploying Scalable Agents 的 OTel 实践#

MAF 原生集成 OpenTelemetry——每个 LLM 调用、工具调用、编排步骤都是 span：

from agent_framework.observability import get_tracer

tracer = get_tracer()

with tracer.start_as_current_span("support_request") as span:
    span.set_attribute("customer.tier", "enterprise")
    span.set_attribute("routed.model", "gpt-5-nano")
    # agent 调用自动产生子 span
    response = await support_agent.run(query)

Span attributes 把 trace 变成可搜索的问题：

• "企业客户被路由到小模型的频率？"

• "p95 延迟是否在 SLA 内？"

• "退款工具的失败率？"

八、成本优化：Deploying Scalable Agents 的三大杠杆#

成本 = token 花费。三个杠杆按影响力排序：

1. 选择合适的模型：通过评估门的小模型永远比大模型便宜

2. 按复杂度路由：复杂请求才付大模型的钱

3. 激进缓存：最便宜的模型调用就是你不发的那次

"Evaluation tells you the quality floor; routing and caching keep you as close to that floor's cost as possible."

九、企业级关注点#

9.1 治理（Governance）#

Hosted Agents 继承 Foundry 的 RBAC、内容安全、审计日志。

# 给 Agent 分配托管身份 + 最小权限
az role assignment create \
  --assignee <agent-managed-identity> \
  --role "Storage Blob Data Reader" \
  --scope /subscriptions/.../storageAccounts/...

9.2 人审（HITL）#

高风险动作必须人工审批——退款、删账户、升级到法务：

@tool(approval_mode="require")  # 标记需要人审
def issue_refund(order_id: str, amount: float) -> str:
    """Issue a refund for an order."""
    # 实际执行前会暂停 workflow 等人批
    return process_refund(order_id, amount)

9.3 MCP 的生产安全#

把每个 MCP Server 当成不可信边界：

• 固定版本（防恶意升级）

• 限定身份（用 scoped identity）

• 验证输出（不直接信任 MCP 返回）

• 不暴露密钥（密钥不进 MCP context）

十、动手：评估门（Evaluation Gate）#

Deploying Scalable Agents 给了一个评估门示例：

async def evaluation_gate(agent, test_cases, threshold: float = 0.8) -> bool:
    """评估 Agent 是否满足质量门槛"""
    passed = 0
    for case in test_cases:
        result = await agent.run(case["input"])
        if score_response(result.text, case["expected"]) >= 0.8:
            passed += 1

    pass_rate = passed / len(test_cases)
    print(f"评估通过率：{pass_rate:.0%}（门槛：{threshold:.0%}）")

    return pass_rate >= threshold  # 只有通过才能部署

CI/CD 集成：

# .github/workflows/deploy-agent.yml
- name: 评估 Agent
  run: python evaluation_gate.py
- name: 部署 Agent
  if: success()
  run: azd deploy

十一、冒烟测试（Smoke Test）#

Deploying Scalable Agents 还提到了一个轻量级的部署后验证：

"Deploying 'successfully' only proves the control plane accepted the definition — it does not prove the agent responds."

# .github/workflows/smoke-test.yml
- name: 冒烟测试 Agent
  uses: JFolberth/ai-smoketest@v1
  with:
    project_endpoint: ${{ secrets.PROJECT_ENDPOINT }}
    agent_name: ContosoSupportAgent
    tests_file: tests/lesson-16-smoke-tests.json

分层验证金字塔：

┌─────────────────────────┐
│  在线评估（持续）        │  ← 真实流量
├─────────────────────────┤
│  离线评估（部署前）      │  ← 评估门
├─────────────────────────┤
│  冒烟测试（每次部署）    │  ← 能响应？
└─────────────────────────┘

十二、关键要点回顾#

核心原则：离线评估是部署的门禁，不是事后补的功课。

十三、参考资源#

• 课程章节：Deploying Scalable Agents

• Microsoft Foundry：https://learn.microsoft.com/azure/ai-foundry/what-is-azure-ai-foundry

• Agent Service：https://learn.microsoft.com/azure/ai-foundry/agents/overview

• MAF：https://learn.microsoft.com/en-us/agent-framework/overview/

• OpenTelemetry：https://opentelemetry.io/

十四、下一课预告#

L18 本地 Agent：把 Agent 跑在你自己的机器上——不联网、零 API 成本、隐私安全。

为什么重要：很多场景（医疗、法律、内部数据）不能上云——本地 Agent 是刚需。