概述

概述

使用 Foundry Local 将 AI 模型视为模块化、可定制的工具，直接在设备上运行。本次课程重点介绍隐私保护、低延迟推理的实用工作流程，以及如何通过 SDK、API 或 CLI 集成这些工具。您还将学习如何在需要时扩展到 Azure AI Foundry。

** 针对现代 SDK 更新**：本模块已与最新的 Microsoft Foundry-Local 仓库模式对齐，并匹配 samples/06/ 中的智能路由实现。示例现已使用现代 foundry-local-sdk 和高级模型选择策略。

️ 架构亮点：

• 智能模型路由：基于关键词选择通用、推理、代码和创意模型
• 现代 SDK 集成：使用 FoundryLocalManager 进行自动服务发现
• 环境配置：通过环境变量灵活分配模型
• 健康监控：服务验证和模型可用性检查
• 生产就绪：全面的错误处理和备用机制

** 本地实现：**

• samples/06/router.py - 基于关键词选择的智能模型路由器
• samples/06/model_router.ipynb - 交互式示例和基准测试
• samples/06/README.md - 配置和使用说明

参考资料：

• Foundry Local 文档：https://learn.microsoft.com/en-us/azure/ai-foundry/foundry-local/
• 集成推理 SDK：https://learn.microsoft.com/en-us/azure/ai-foundry/foundry-local/how-to/how-to-integrate-with-inference-sdks
• 编译 Hugging Face 模型：https://learn.microsoft.com/en-us/azure/ai-foundry/foundry-local/how-to/how-to-compile-hugging-face-models

概述

使用 Foundry Local 将 AI 模型视为模块化、可定制的工具，直接在设备上运行。本次课程重点介绍隐私保护、低延迟推理的实用工作流程，以及如何通过 SDK、API 或 CLI 集成这些工具。您还将学习如何在需要时扩展到 Azure AI Foundry。

参考资料：

• Foundry Local 文档：https://learn.microsoft.com/en-us/azure/ai-foundry/foundry-local/
• 集成推理 SDK：https://learn.microsoft.com/en-us/azure/ai-foundry/foundry-local/how-to/how-to-integrate-with-inference-sdks
• 编译 Hugging Face 模型：https://learn.microsoft.com/en-us/azure/ai-foundry/foundry-local/how-to/how-to-compile-hugging-face-models

学习目标

• 设计设备上的模型工具模式
• 通过 OpenAI 兼容的 REST API 或 SDK 集成
• 根据领域需求定制模型
• 规划向 Azure AI Foundry 的混合扩展

第 1 部分：智能模型路由器（现代实现）

目标：通过基于查询内容的自动路由实现智能模型选择。

** 注意**：此实现与 samples/06/router.py 中使用的模式相匹配，采用高级关键词模型选择。

步骤 1）使用 FoundryLocalManager 定义现代模型路由器

# router/intelligent_router.py
from foundry_local import FoundryLocalManager
from openai import OpenAI
from typing import Dict, Any, Optional
import os
import json

class ModelRouter:
    """Intelligent model router that selects appropriate models for different task types."""
    
    def __init__(self):
        self.client = None
        self.base_url = None
        self.tools = self._load_tool_registry()
        self._initialize_client()
    
    def _load_tool_registry(self) -> Dict[str, Dict[str, Any]]:
        """Load tool registry from environment or use defaults."""
        default_tools = {
            "general": {
                "model": os.environ.get("GENERAL_MODEL", "phi-4-mini"),
                "notes": "Fast general-purpose chat and Q&A",
                "temperature": 0.7
            },
            "reasoning": {
                "model": os.environ.get("REASONING_MODEL", "deepseek-r1-7b"),
                "notes": "Step-by-step analysis and logical reasoning",
                "temperature": 0.3
            },
            "code": {
                "model": os.environ.get("CODE_MODEL", "qwen2.5-7b"),
                "notes": "Code generation, debugging, and technical tasks",
                "temperature": 0.2
            },
            "creative": {
                "model": os.environ.get("CREATIVE_MODEL", "phi-4-mini"),
                "notes": "Creative writing and storytelling",
                "temperature": 0.9
            }
        }
        
        # Check for environment override
        tools_env = os.environ.get("TOOL_REGISTRY")
        if tools_env:
            try:
                return json.loads(tools_env)
            except json.JSONDecodeError:
                print("Warning: Invalid TOOL_REGISTRY JSON, using defaults")
        
        return default_tools
​

步骤 2）使用现代 SDK 和服务发现初始化客户端

    def _initialize_client(self):
        """Initialize OpenAI client with Foundry Local or fallback configuration."""
        try:
            from foundry_local import FoundryLocalManager
            # Try to use any available model for client initialization
            first_model = next(iter(self.tools.values()))["model"]
            manager = FoundryLocalManager(first_model)
            
            self.client = OpenAI(
                base_url=manager.endpoint,
                api_key=manager.api_key
            )
            self.base_url = manager.endpoint
            print(f"✅ Foundry Local SDK initialized")
        except Exception as e:
            print(f"Warning: Could not use Foundry SDK ({e}), falling back to manual configuration")
            # Fallback to manual configuration
            self.base_url = os.environ.get("BASE_URL", "http://localhost:8000")
            api_key = os.environ.get("API_KEY", "")
            
            self.client = OpenAI(
                base_url=f"{self.base_url}/v1",
                api_key=api_key
            )
            print(f"Initialized manual configuration at {self.base_url}")
    
    def select_tool(self, user_query: str) -> str:
        """Select the most appropriate tool based on the user query."""
        query_lower = user_query.lower()
        
        # Code-related keywords
        code_keywords = ["code", "python", "function", "class", "method", "bug", "debug", 
                        "programming", "script", "algorithm", "implementation", "refactor"]
        if any(keyword in query_lower for keyword in code_keywords):
            return "code"
        
        # Reasoning keywords
        reasoning_keywords = ["why", "how", "explain", "step-by-step", "reason", "analyze", 
                             "think", "logic", "because", "cause", "compare", "evaluate"]
        if any(keyword in query_lower for keyword in reasoning_keywords):
            return "reasoning"
        
        # Creative keywords
        creative_keywords = ["story", "poem", "creative", "imagine", "write", "tale", 
                           "narrative", "fiction", "character", "plot"]
        if any(keyword in query_lower for keyword in creative_keywords):
            return "creative"
        
        # Default to general
        return "general"
    
    def chat(self, model: str, content: str, max_tokens: int = 300, temperature: Optional[float] = None) -> str:
        """Send chat completion request to the specified model."""
        try:
            params = {
                "model": model,
                "messages": [{"role": "user", "content": content}],
                "max_tokens": max_tokens
            }
            
            if temperature is not None:
                params["temperature"] = temperature
            
            response = self.client.chat.completions.create(**params)
            return response.choices[0].message.content
        except Exception as e:
            return f"Error generating response with model {model}: {str(e)}"
​

步骤 3）实现智能路由和执行（参见 samples/06/router.py）

    def route_and_run(self, prompt: str) -> Dict[str, Any]:
        """Route the prompt to the appropriate model and generate response."""
        tool_key = self.select_tool(prompt)
        tool_config = self.tools[tool_key]
        model = tool_config["model"]
        temperature = tool_config.get("temperature", 0.7)
        
        print(f" Selected tool: {tool_key} (model: {model})")
        
        answer = self.chat(
            model=model, 
            content=prompt, 
            max_tokens=400, 
            temperature=temperature
        )
        
        return {
            "tool": tool_key,
            "model": model,
            "tool_description": tool_config["notes"],
            "temperature": temperature,
            "answer": answer
        }
    
    def check_service_health(self) -> Dict[str, Any]:
        """Check Foundry Local service health and available models."""
        try:
            models_response = self.client.models.list()
            available_models = [model.id for model in models_response.data]
            
            return {
                "status": "healthy",
                "base_url": self.base_url,
                "available_models": available_models,
                "tools_configured": list(self.tools.keys())
            }
        except Exception as e:
            return {
                "status": "error",
                "base_url": self.base_url,
                "error": str(e)
            }

if __name__ == "__main__":
    # Ensure: foundry model run phi-4-mini
    router = ModelRouter()
    
    # Check health
    health = router.check_service_health()
    print(f"Service Health: {json.dumps(health, indent=2)}")
    
    # Test different query types
    queries = [
        "Write a Python function to calculate fibonacci numbers",  # -> code
        "Explain step-by-step why the sky is blue",  # -> reasoning
        "Tell me a creative story about AI",  # -> creative
        "What's the weather like today?"  # -> general
    ]
    
    for query in queries:
        result = router.route_and_run(query)
        print(f"\nQuery: {query}")
        print(f"Selected: {result['tool']} -> {result['model']}")
        print(f"Answer: {result['answer'][:100]}...")
​

第 2 部分：现代 SDK 集成（逐步指南）

目标：使用 Foundry Local SDK 与 OpenAI Python SDK 实现无缝集成。

步骤 1）安装依赖项

cd Module08
.\.venv\Scripts\activate
pip install foundry-local-sdk openai
​

步骤 2）配置环境（可选 - 参见 samples/06/README.md）

REM Override default models per tool
set GENERAL_MODEL=phi-4-mini
set REASONING_MODEL=deepseek-r1-7b
set CODE_MODEL=qwen2.5-7b
REM Or provide a full JSON registry
set TOOL_REGISTRY={"general":{"model":"phi-4-mini"},"reasoning":{"model":"deepseek-r1-7b"}}
​

步骤 3）现代 SDK 集成

# modern_sdk_demo.py
from foundry_local import FoundryLocalManager
from openai import OpenAI
import sys

def main():
    """Demonstrate modern SDK integration."""
    try:
        # Initialize with FoundryLocalManager
        alias = "phi-4-mini"
        manager = FoundryLocalManager(alias)
        
        # Create OpenAI client using Foundry Local endpoint
        client = OpenAI(
            base_url=manager.endpoint,
            api_key=manager.api_key
        )
        
        # Get model info
        model_info = manager.get_model_info(alias)
        print(f"Using model: {model_info.id}")
        
        # Make request with streaming
        stream = client.chat.completions.create(
            model=model_info.id,
            messages=[{"role": "user", "content": "Explain edge AI benefits in one paragraph."}],
            stream=True,
            max_tokens=200
        )
        
        print("Response: ", end="")
        for chunk in stream:
            if chunk.choices[0].delta.content:
                print(chunk.choices[0].delta.content, end="", flush=True)
        print()
        
    except Exception as e:
        print(f"Error: {e}")
        print("Ensure Foundry Local is running with: foundry model run phi-4-mini")
        sys.exit(1)

if __name__ == "__main__":
    main()
​

第 3 部分：领域定制（逐步指南）

目标：使用提示模板和 JSON 模式为特定领域定制输出。

步骤 1）创建领域提示模板

# domain/templates.py
BUSINESS_ANALYST_SYSTEM = """
You are a senior business analyst. Provide:
1) Key insights
2) Risks
3) Next steps
Respond in valid JSON with fields: insights, risks, next_steps.
"""
​

步骤 2）强制输出 JSON 格式

# domain/analyst.py
import requests, os, json

BASE_URL = os.getenv("OPENAI_BASE_URL", "http://localhost:8000/v1")
API_KEY = os.getenv("OPENAI_API_KEY", "local-key")
HEADERS = {"Content-Type":"application/json","Authorization":f"Bearer {API_KEY}"}

from domain.templates import BUSINESS_ANALYST_SYSTEM

def analyze(text: str) -> dict:
    messages = [
        {"role":"system","content": BUSINESS_ANALYST_SYSTEM},
        {"role":"user","content": f"Analyze this business text:\n{text}"}
    ]
    r = requests.post(f"{BASE_URL}/chat/completions", json={
    "model":"phi-4-mini",
        "messages": messages,
        "response_format": {"type":"json_object"},
        "temperature": 0.3
    }, headers=HEADERS, timeout=60)
    r.raise_for_status()
    # Parse JSON content
    content = r.json()["choices"][0]["message"]["content"]
    return json.loads(content)

if __name__ == "__main__":
    print(analyze("Sales dipped 12% in Q3 due to supply constraints and marketing cuts."))
​

第 4 部分：离线和安全姿态（逐步指南）

目标：在本地运行模型工具时确保隐私和弹性。

步骤 1）预热并验证本地端点

foundry model run phi-4-mini
curl http://localhost:8000/v1/models
​

步骤 2）清理输入

# security/sanitize.py
import re
EMAIL_RE = re.compile(r"[\w\.-]+@[\w\.-]+")
PHONE_RE = re.compile(r"\+?\d[\d\s\-]{7,}\d")

def sanitize(text: str) -> str:
    text = EMAIL_RE.sub("[REDACTED_EMAIL]", text)
    text = PHONE_RE.sub("[REDACTED_PHONE]", text)
    return text
​

步骤 3）启用本地专用标志和日志记录

# security/local_only.py
import os, json, time
LOG = os.getenv("MODELS_AS_TOOLS_LOG", "./tools_logs.jsonl")

def record(event: dict):
    with open(LOG, "a", encoding="utf-8") as f:
        f.write(json.dumps(event) + "\n")

# Usage before each call
def before_call(tool_name, payload):
    record({"ts": time.time(), "tool": tool_name, "event": "before_call"})

# After each call
def after_call(tool_name, result):
    record({"ts": time.time(), "tool": tool_name, "event": "after_call"})
​

第 5 部分：生产部署和扩展

目标：通过监控和 Azure AI Foundry 集成部署智能路由器。

** 注意**：samples/06/model_router.ipynb 中的本地实现包含生产部署模式的全面示例。

步骤 1）带监控的生产路由器（参见 samples/06/router.py）

# production/router.py
from router.intelligent_router import ModelRouter
import json
import time
import sys

class ProductionModelRouter(ModelRouter):
    """Production-ready model router with monitoring and logging."""
    
    def __init__(self):
        super().__init__()
        self.request_count = 0
        self.error_count = 0
        self.start_time = time.time()
    
    def route_and_run_with_monitoring(self, prompt: str) -> Dict[str, Any]:
        """Route with comprehensive monitoring and error handling."""
        start_time = time.time()
        self.request_count += 1
        
        try:
            result = self.route_and_run(prompt)
            processing_time = time.time() - start_time
            
            # Log successful request
            self._log_request({
                "status": "success",
                "tool": result["tool"],
                "model": result["model"],
                "processing_time": processing_time,
                "timestamp": time.strftime("%Y-%m-%d %H:%M:%S")
            })
            
            result["processing_time"] = processing_time
            return result
            
        except Exception as e:
            self.error_count += 1
            error_result = {
                "status": "error",
                "error": str(e),
                "processing_time": time.time() - start_time,
                "timestamp": time.strftime("%Y-%m-%d %H:%M:%S")
            }
            
            self._log_request(error_result)
            return error_result
    
    def _log_request(self, data: Dict[str, Any]):
        """Log request data for monitoring."""
        print(f" {json.dumps(data)}")
    
    def get_stats(self) -> Dict[str, Any]:
        """Get router statistics."""
        uptime = time.time() - self.start_time
        return {
            "uptime_seconds": uptime,
            "total_requests": self.request_count,
            "error_count": self.error_count,
            "success_rate": (self.request_count - self.error_count) / max(1, self.request_count),
            "requests_per_minute": self.request_count / max(1, uptime / 60)
        }

def main():
    """Production router demo."""
    router = ProductionModelRouter()
    
    # Health check
    health = router.check_service_health()
    if health["status"] == "error":
        print(f"❌ Service health check failed: {health['error']}")
        sys.exit(1)
    
    print(f"✅ Service healthy with {len(health['available_models'])} models")
    
    # Process user query
    user_prompt = " ".join(sys.argv[1:]) or "Write three benefits of on-device AI in JSON format."
    print(f"\n Processing: {user_prompt}")
    
    result = router.route_and_run_with_monitoring(user_prompt)
    
    if result.get("status") == "error":
        print(f"❌ Error: {result['error']}")
    else:
        print(f"\n Result:")
        print(f"Tool: {result['tool']} -> Model: {result['model']}")
        print(f"Processing Time: {result['processing_time']:.2f}s")
        print(f"Answer: {result['answer']}")
    
    # Show stats
    stats = router.get_stats()
    print(f"\n Statistics: {json.dumps(stats, indent=2)}")

if __name__ == "__main__":
    main()
​

实践清单

• 使用关键词选择实现智能模型路由器（samples/06/router.py）
• 配置多个专用模型（通用、推理、代码、创意）
• 测试交互式 Jupyter notebook（samples/06/model_router.ipynb）
• 设置基于环境的模型配置
• 实现服务健康监控和错误处理
• 部署带全面日志记录的生产路由器

本地示例集成

运行完整实现：

cd Module08
.\.venv\Scripts\activate

REM Start required models
foundry model run phi-4-mini
foundry model run qwen2.5-7b
foundry model run deepseek-r1-7b

REM Test the intelligent router
python samples\06\router.py "Write a Python function to sort a list"
python samples\06\router.py "Explain step-by-step how bubble sort works"
python samples\06\router.py "Tell me a creative story about robots"

REM Explore the interactive notebook
jupyter notebook samples/06/model_router.ipynb
​

参考资料和后续步骤

• 本地实现：samples/06/ - 支持多模型的完整智能路由器
• Microsoft 示例：Hello Foundry Local
• 集成文档：集成推理 SDK
• 高级模式：探索模块 5 中的函数调用和多代理编排

总结

Foundry Local 使得在设备上运行的 AI 模型成为智能、专用的工具。通过自动模型选择、全面监控和生产就绪模式，团队可以交付适应不同任务类型的复杂 AI 应用，同时保持隐私和性能。本次课程展示的智能路由器模式为构建复杂 AI 系统提供了基础，这些系统可以从本地开发扩展到生产部署。

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