第六节：FoundryLocal–模型作为工具

第六节：Foundry Local – 模型作为工具

摘要

将模型视为本地 AI 操作层中的可组合工具。本节课展示如何串联多个专用 SLM/LLM 调用，选择性地路由任务，并向应用程序公开统一的 SDK 接口。您将构建一个轻量级模型路由器和任务规划器，将其集成到应用脚本中，并概述向 Azure AI Foundry 扩展以支持生产工作负载的路径。

学习目标

• 概念化模型为具有声明能力的原子工具
• 路由基于意图/启发式评分的请求
• 串联跨多步骤任务的输出（分解 → 解决 → 精炼）
• 集成统一的客户端 API 以供下游应用使用
• 扩展设计到云端（与 OpenAI 兼容的合同）

前置条件

• 完成第 1–5 节
• 缓存多个本地模型（例如，phi-4-mini，deepseek-coder-1.3b，qwen2.5-0.5b）

跨平台环境代码片段

Windows PowerShell:

py -m venv .venv
 .\.venv\Scripts\Activate.ps1
pip install --upgrade pip
pip install foundry-local-sdk openai
​

macOS / Linux:

python3 -m venv .venv
source .venv/bin/activate
python -m pip install --upgrade pip
pip install foundry-local-sdk openai
​

从 macOS 远程/虚拟机服务访问：

export FOUNDRY_LOCAL_ENDPOINT=http://<windows-host>:5273/v1
​

演示流程（30 分钟）

1. 工具能力声明（5 分钟）

创建 samples/06-tools/models_catalog.py：

CATALOG = {
  "phi-4-mini": {
    "capabilities": ["general", "reasoning", "summarize"],
    "priority": 2
  },
  "deepseek-coder-1.3b": {
    "capabilities": ["code", "refactor", "explain_code"],
    "priority": 1
  },
  "qwen2.5-0.5b": {
    "capabilities": ["fast", "classification", "lightweight"],
    "priority": 3
  }
}
​

2. 意图检测与路由（8 分钟）

创建 samples/06-tools/router.py：

#!/usr/bin/env python3
"""Model-as-tool router using Foundry Local OpenAI-compatible endpoint."""
from openai import OpenAI
from models_catalog import CATALOG
import re

client = OpenAI(base_url="http://localhost:5273/v1", api_key="not-needed")

INTENT_RULES = [
  (re.compile(r"code|function|refactor|bug|optimi", re.I), "code"),
  (re.compile(r"summari|abstract|tl;dr", re.I), "summarize"),
  (re.compile(r"classif|label|category", re.I), "classification"),
]

def detect_intent(prompt: str) -> str:
    for pat, intent in INTENT_RULES:
        if pat.search(prompt):
            return intent
    return "general"

def select_model(intent: str) -> str:
    # Score catalog: capability match first, then priority
    scored = []
    for name, meta in CATALOG.items():
        caps = meta["capabilities"]
        match = intent in caps
        scored.append((name, match, meta["priority"]))
    # Sort: match True first, then lowest priority value
    scored.sort(key=lambda t: (not t[1], t[2]))
    return scored[0][0]

def run(prompt: str):
    intent = detect_intent(prompt)
    model = select_model(intent)
    resp = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        max_tokens=400,
        temperature=0.5
    )
    return {"intent": intent, "model": model, "output": resp.choices[0].message.content}

if __name__ == "__main__":
    tests = [
        "Refactor this Python function for readability",
        "Summarize the importance of local AI governance",
        "Classify this feedback: 'The UI is slow and confusing'"
    ]
    for t in tests:
        r = run(t)
        print(f"Prompt: {t}\nModel: {r['model']} (intent={r['intent']})\nOutput: {r['output'][:160]}...\n")
​

3. 多步骤任务串联（7 分钟）

创建 samples/06-tools/pipeline.py：

#!/usr/bin/env python3
"""Multi-step pipeline: plan -> solve -> refine using specialized models."""
from openai import OpenAI
from router import detect_intent, select_model

client = OpenAI(base_url="http://localhost:5273/v1", api_key="not-needed")

def chat(model, content, temp=0.4):
    r = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": content}],
        max_tokens=350,
        temperature=temp
    )
    return r.choices[0].message.content

def pipeline(task: str):
    plan_model = select_model("general")
    plan = chat(plan_model, f"Break the task into 3 ordered steps. Task: {task}")
    steps = [s for s in plan.split('\n') if s.strip()][:3]
    outputs = []
    for step in steps:
        intent = detect_intent(step)
        model = select_model(intent)
        out = chat(model, step)
        outputs.append((step, model, out))
    refine_model = select_model("summarize")
    combined = '\n'.join(o[2] for o in outputs)
    refined = chat(refine_model, f"Condense results into a cohesive answer:\n{combined}")
    return {"plan": plan, "steps": outputs, "final": refined}

if __name__ == '__main__':
    result = pipeline("Generate a refactored version of a slow Python loop and summarize performance gains.")
    print("PLAN:\n", result['plan'])
    print("FINAL:\n", result['final'][:400])
​

4. 初始项目：改进 06-models-as-tools（5 分钟）

增强功能：

• 添加流式令牌支持（渐进式 UI 更新）
• 添加置信评分：词汇重叠或提示规则
• 导出追踪 JSON（意图 → 模型 → 延迟 → 令牌使用）
• 实现缓存重用以支持重复的子步骤

5. 向 Azure 扩展路径（5 分钟）

验证检查表

foundry model run phi-4-mini
foundry model run deepseek-coder-1.3b
python samples/06-tools/router.py
python samples/06-tools/pipeline.py
​

预期基于意图的模型选择和最终精炼输出。

故障排除

参考资料

• Foundry Local SDK: https://github.com/microsoft/Foundry-Local/tree/main/sdk/python
• Azure AI Foundry 文档: https://learn.microsoft.com/azure/ai-foundry
• 提示质量模式：参见第 2 节

课程时长：30 分钟
难度：专家级

示例场景与工作坊映射

场景叙述

一个工程生产力工具接收异构任务：代码重构、总结架构笔记、分类反馈。为了最小化延迟和资源使用，一个小型通用模型负责规划和总结，一个专门的代码模型处理重构，一个轻量级分类模型标记反馈。管道脚本展示了串联和精炼；路由脚本隔离了自适应单提示路由。

目录快照

CATALOG = {
    "phi-4-mini": {"capabilities": ["general", "summarize"], "priority": 2},
    "deepseek-coder-1.3b": {"capabilities": ["code", "refactor"], "priority": 1},
    "qwen2.5-0.5b": {"capabilities": ["classification", "fast"], "priority": 3}
}
​

示例测试提示

[
    "Refactor this Python function for readability",
    "Summarize the importance of small language models",
    "Classify this feedback: The UI is slow but pretty",
    "Generate a refactored version of a slow Python loop and summarize performance gains."
]
​

追踪扩展（可选）

为 models_pipeline.py 添加每步骤追踪 JSON 行：

trace.append({
    "step": step_idx,
    "intent": intent,
    "alias": alias,
    "latency_ms": round((end-start)*1000,2),
    "tokens": getattr(usage,'total_tokens',None)
})
​

升级启发式（想法）

如果规划包含关键词如“优化”、“安全”或步骤长度 > 280 字符 → 仅对该步骤升级到更大的模型（例如，gpt-oss-20b）。

可选增强功能

追踪结构示例

trace.append({
  "step": idx,
  "intent": intent,
  "alias": alias,
  "latency_ms": round((end-start)*1000,2),
  "tokens": getattr(usage,'total_tokens',None)
})
​

自适应升级草图

if len(prompt) > 280 or 'compliance' in prompt.lower():
    # escalate to larger reasoning model if available
    alias = 'gpt-oss-20b'
​

模型目录热加载

import json, time, os
CATALOG_PATH = 'catalog.json'
last_mtime = 0
def get_catalog():
    global last_mtime, CATALOG
    m = os.path.getmtime(CATALOG_PATH)
    if m != last_mtime:
        CATALOG = json.load(open(CATALOG_PATH))
        last_mtime = m
    return CATALOG
​

免责声明：
本文档使用AI翻译服务Co-op Translator进行翻译。尽管我们努力确保翻译的准确性，但请注意，自动翻译可能包含错误或不准确之处。原始语言的文档应被视为权威来源。对于重要信息，建议使用专业人工翻译。我们不对因使用此翻译而产生的任何误解或误读承担责任。