06.ModelsAsTools.md

概述

使用 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 系統提供了基礎，這些系統可以從本地開發擴展到生產部署。

---

免責聲明：
本文件已使用人工智能翻譯服務 Co-op Translator 進行翻譯。儘管我們致力於提供準確的翻譯，但請注意，自動翻譯可能包含錯誤或不準確之處。原始文件的母語版本應被視為權威來源。對於重要信息，建議使用專業人工翻譯。我們對因使用此翻譯而引起的任何誤解或錯誤解釋概不負責。