第五节：使用FoundryLocal快速构建AI驱动的代理

第五节：使用 Foundry Local 快速构建 AI 驱动的代理

注意：Foundry Local 中的代理功能会不断演进——在实施高级模式之前，请确认最新版本说明中的支持情况。

概述

使用 Foundry Local 快速原型化代理应用程序：包括系统提示、数据基础和编排模式。当代理支持可用时，可以标准化使用与 OpenAI 兼容的函数调用，或在混合设计中使用 Azure AI Agents 的云端功能。

** 适配现代 SDK**：本模块已与最新的 Microsoft Foundry-Local 仓库模式对齐，并匹配 samples/05/ 中的全面实现。示例现已使用现代 foundry-local-sdk 和 OpenAI 客户端，而非手动请求。

️ 架构亮点：

• 专业代理：检索、推理和执行代理，具备独特能力
• 协调器模式：通过反馈循环编排多代理工作流
• 现代 SDK 集成：使用 FoundryLocalManager 和 OpenAI 客户端
• 生产就绪：包括错误处理、性能监控和健康检查
• 全面示例：带有高级功能的交互式 Jupyter notebook

** 本地实现：**

• samples/05/multi_agent_orchestration.ipynb - 交互式示例和基准测试
• samples/05/agents/specialists.py - 代理实现
• samples/05/agents/coordinator.py - 编排逻辑

参考资料：

• Foundry Local 文档：https://learn.microsoft.com/en-us/azure/ai-foundry/foundry-local/
• Azure AI Foundry Agents：https://learn.microsoft.com/en-us/azure/ai-services/agents/overview
• 函数调用示例（Foundry Local 示例）：https://github.com/microsoft/Foundry-Local/tree/main/samples/python/functioncalling

学习目标

• 设计系统提示和数据基础策略以确保可靠行为
• 实现函数调用（工具使用）模式
• 编排多代理工作流（本地和混合）
• 规划可观察性和安全性

第一部分：系统提示和数据基础

• 定义严格的角色、约束和输出模式
• 使用本地或企业数据作为响应基础
• 强制 JSON 输出以支持后续自动化

第二部分：函数调用（现代 SDK 方法）

# tools.py
import json
from typing import List, Dict, Any

def get_weather(city: str) -> str:
    return f"Weather in {city}: Sunny, 25C"

# Modern tools format for OpenAI API
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Get current weather for a city",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "City name"}
                },
                "required": ["city"]
            }
        }
    }
]
​

# agent.py
from foundry_local import FoundryLocalManager
from openai import OpenAI
import json
from tools import TOOLS, get_weather

# Initialize Foundry Local Manager
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
)

SYSTEM_PROMPT = "You are a helpful assistant. Use tools when needed."

def process_function_call(messages: List[Dict], tools: List[Dict]) -> str:
    """Process function calling with modern OpenAI API."""
    try:
        response = client.chat.completions.create(
            model=manager.get_model_info(alias).id,
            messages=messages,
            tools=tools,
            tool_choice="auto"
        )
        
        message = response.choices[0].message
        
        if message.tool_calls:
            # Handle function calls
            messages.append(message)
            
            for tool_call in message.tool_calls:
                if tool_call.function.name == "get_weather":
                    args = json.loads(tool_call.function.arguments)
                    result = get_weather(args["city"])
                    
                    # Add function result to messages
                    messages.append({
                        "role": "tool",
                        "tool_call_id": tool_call.id,
                        "content": result
                    })
            
            # Get final response
            final_response = client.chat.completions.create(
                model=manager.get_model_info(alias).id,
                messages=messages
            )
            return final_response.choices[0].message.content
        else:
            return message.content
            
    except Exception as e:
        return f"Error: {str(e)}"

# Example usage
messages = [
    {"role": "system", "content": SYSTEM_PROMPT},
    {"role": "user", "content": "What's the weather in Paris?"}
]

result = process_function_call(messages, TOOLS)
print(result)
​

运行：

# Ensure Foundry Local is running with a model
foundry model run phi-4-mini
python agent.py
​

第三部分：多代理编排（模式）

设计一个协调器，通过 Foundry Local 的 OpenAI 兼容端点将任务分配给专业代理（检索、推理、执行）。

步骤 1）使用现代 SDK 定义专业代理（参见 samples/05/agents/specialists.py）

# agents/specialists.py
from foundry_local import FoundryLocalManager
from openai import OpenAI
from typing import List, Dict, Any

class FoundryClient:
    """Shared client for all specialist agents."""
    
    def __init__(self, model_alias: str = "phi-4-mini"):
        self.client = None
        self.model_name = None
        self.model_alias = model_alias
        self._initialize_client()
    
    def _initialize_client(self):
        """Initialize OpenAI client with Foundry Local."""
        try:
            manager = FoundryLocalManager(self.model_alias)
            model_info = manager.get_model_info(self.model_alias)
            
            self.client = OpenAI(
                base_url=manager.endpoint,
                api_key=manager.api_key
            )
            self.model_name = model_info.id
            print(f"✅ Foundry Local initialized with model: {self.model_name}")
        except Exception as e:
            print(f"❌ Error initializing Foundry Local: {e}")
            raise
    
    def chat(self, messages: List[Dict[str, str]], max_tokens: int = 300, temperature: float = 0.4) -> str:
        """Send chat completion request to the model."""
        try:
            response = self.client.chat.completions.create(
                model=self.model_name,
                messages=messages,
                max_tokens=max_tokens,
                temperature=temperature
            )
            return response.choices[0].message.content
        except Exception as e:
            return f"Error generating response: {str(e)}"

# Global client instance
_client = FoundryClient()

class RetrievalAgent:
    """Agent specialized in retrieving relevant information from knowledge sources."""
    
    SYSTEM = """You are a specialized retrieval agent. Your job is to extract and retrieve 
    the most relevant information from knowledge sources based on a given query. Focus on key facts, 
    data points, and contextual information that would be useful for decision-making."""
    
    def run(self, query: str) -> str:
        """Retrieve relevant information based on the query."""
        messages = [
            {"role": "system", "content": self.SYSTEM},
            {"role": "user", "content": f"Query: {query}\n\nRetrieve the most relevant key facts, data points, and contextual information that would help answer this query or support decision-making around it."}
        ]
        return _client.chat(messages)

class ReasoningAgent:
    """Agent specialized in step-by-step analysis and reasoning."""
    
    SYSTEM = """You are a specialized reasoning agent. Your job is to analyze inputs 
    step-by-step and produce structured, logical conclusions. Break down complex problems 
    into manageable parts and provide clear reasoning for your conclusions."""
    
    def run(self, context: str, question: str) -> str:
        """Analyze context and question to produce structured conclusions."""
        messages = [
            {"role": "system", "content": self.SYSTEM},
            {"role": "user", "content": f"Context:\n{context}\n\nQuestion: {question}\n\nAnalyze this step-by-step and provide a structured, logical conclusion with clear reasoning."}
        ]
        return _client.chat(messages, max_tokens=400)

class ExecutionAgent:
    """Agent specialized in creating actionable execution plans."""
    
    SYSTEM = """You are a specialized execution agent. Your job is to transform decisions 
    and conclusions into concrete, actionable steps. Always format your response as valid JSON 
    with an array of action items. Each action should be specific, measurable, and achievable."""
    
    def run(self, decision: str) -> str:
        """Transform decision into actionable steps in JSON format."""
        messages = [
            {"role": "system", "content": self.SYSTEM},
            {"role": "user", "content": f"Decision/Conclusion:\n{decision}\n\nCreate 3-5 specific, actionable steps to implement this decision. Format as JSON with this structure:\n{{\"actions\": [{{\"step\": 1, \"description\": \"...\", \"priority\": \"high/medium/low\", \"timeline\": \"...\"}}]}}"}
        ]
        return _client.chat(messages, max_tokens=400, temperature=0.3)
​

步骤 2）构建具有高级功能的协调器

# agents/coordinator.py
from .specialists import RetrievalAgent, ReasoningAgent, ExecutionAgent
from typing import Dict, Any
import time
import json

class Coordinator:
    """Multi-agent coordinator that orchestrates specialist agents to handle complex tasks."""
    
    def __init__(self):
        """Initialize the coordinator with specialist agents."""
        self.retrieval = RetrievalAgent()
        self.reasoning = ReasoningAgent()
        self.execution = ExecutionAgent()
    
    def handle(self, user_goal: str) -> Dict[str, Any]:
        """
        Orchestrate multiple agents to handle a complex user goal.
        
        Args:
            user_goal: The user's high-level goal or request
            
        Returns:
            Dictionary containing the goal, context, decision, and actions
        """
        print(f" **Coordinator:** Processing goal: {user_goal}")
        print("=" * 60)
        
        start_time = time.time()
        
        # Step 1: Retrieve relevant context
        print(" **Step 1:** Retrieving context...")
        context = self.retrieval.run(user_goal)
        print(f"   ✅ Context retrieved ({len(context)} chars)")
        
        # Step 2: Analyze and reason about the context
        print(" **Step 2:** Analyzing and reasoning...")
        decision = self.reasoning.run(context, user_goal)
        print(f"   ✅ Analysis completed ({len(decision)} chars)")
        
        # Step 3: Create actionable execution plan
        print("⚡ **Step 3:** Creating execution plan...")
        actions = self.execution.run(decision)
        print(f"   ✅ Execution plan created ({len(actions)} chars)")
        
        end_time = time.time()
        processing_time = end_time - start_time
        
        result = {
            "goal": user_goal,
            "context": context,
            "decision": decision,
            "actions": actions,
            "agent_flow": ["retrieval", "reasoning", "execution"],
            "processing_time": processing_time,
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S")
        }
        
        print(f"✅ **Coordination Complete** (⏱️ {processing_time:.2f}s)")
        return result
    
    def handle_with_feedback(self, user_goal: str, feedback_rounds: int = 1) -> Dict[str, Any]:
        """
        Handle a goal with multiple feedback rounds for refinement.
        
        Args:
            user_goal: The user's high-level goal or request
            feedback_rounds: Number of feedback rounds to perform
            
        Returns:
            Dictionary containing the refined result
        """
        result = self.handle(user_goal)
        
        for round_num in range(feedback_rounds):
            print(f"\n **Feedback Round {round_num + 1}:**")
            print("-" * 40)
            
            # Use reasoning agent to refine the execution plan
            refinement_prompt = f"""
            Original Goal: {user_goal}
            Current Decision: {result['decision']}
            Current Actions: {result['actions']}
            
            Review the above and suggest improvements or refinements to make the execution plan more effective.
            """
            
            refined_decision = self.reasoning.run(result['context'], refinement_prompt)
            refined_actions = self.execution.run(refined_decision)
            
            result['decision'] = refined_decision
            result['actions'] = refined_actions
            result['refinement_rounds'] = round_num + 1
            
            print(f"   ✅ Round {round_num + 1} refinement completed")
        
        return result

def main():
    """Main function demonstrating the multi-agent coordinator."""
    print(" **Multi-Agent Coordinator Demo**")
    print("=" * 50)
    
    # Create coordinator
    coord = Coordinator()
    
    # Example goals
    example_goals = [
        "Create a plan to onboard 5 new customers this month",
        "Develop a strategy to improve team productivity by 20%",
        "Design a customer feedback collection system"
    ]
    
    # Process example with feedback
    goal = example_goals[0]
    print(f" **Processing Goal:** {goal}")
    print("-" * 50)
    
    try:
        # Basic processing
        result = coord.handle(goal)
        
        # With feedback refinement
        refined_result = coord.handle_with_feedback(goal, feedback_rounds=1)
        
        print("\n **Final Result:**")
        print("=" * 50)
        print(f"**Goal:** {refined_result['goal']}")
        print(f"**Processing Time:** {refined_result['processing_time']:.2f}s")
        
        # Try to parse actions as JSON
        try:
            actions_json = json.loads(refined_result['actions'])
            print(f"\n**Formatted Actions:**")
            print(json.dumps(actions_json, indent=2))
        except (json.JSONDecodeError, TypeError):
            print(f"\n**Actions:** {refined_result['actions']}")
            
    except Exception as e:
        print(f"❌ **Error:** {e}")
        print("\nPlease ensure Foundry Local is running with a model loaded.")

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

步骤 3）验证 Foundry Local 并运行示例

REM Confirm the local endpoint and model are available
foundry model list
foundry model run phi-4-mini
curl http://localhost:8000/v1/models

REM Run the coordinator from Module08 directory
cd Module08
python -m samples.05.agents.coordinator

REM Or explore the comprehensive Jupyter notebook
jupyter notebook samples/05/multi_agent_orchestration.ipynb
​

** 本地示例参考：**

• 主要实现：samples/05/agents/specialists.py 和 samples/05/agents/coordinator.py
• 全面示例：samples/05/multi_agent_orchestration.ipynb
• 设置说明：samples/05/README.md

** 相关 Foundry Local 示例：**

• 函数调用示例
• Hello Foundry Local

指南：

• 在代理之间实现重试和超时
• 添加一个小型内存存储（字典）用于会话/线程状态
• 在多次调用时引入速率限制

第四部分：可观察性和安全性

本地跟踪提示、响应和错误，同时在代理栈中强制数据清理。

步骤 1）轻量级请求日志记录（可选）

注意：以下辅助工具默认未包含。如果需要本地 JSON 日志记录用于实验，请创建 infra/obs.py。

# infra/obs.py
import time, json, os
from datetime import datetime

LOG_DIR = os.getenv("FOUNDRY_AGENT_LOG_DIR", "./agent_logs")
os.makedirs(LOG_DIR, exist_ok=True)

def log_event(kind: str, payload: dict):
    ts = datetime.utcnow().strftime("%Y%m%dT%H%M%SZ")
    path = os.path.join(LOG_DIR, f"{ts}_{kind}.json")
    with open(path, "w", encoding="utf-8") as f:
        json.dump(payload, f, ensure_ascii=False, indent=2)
​

将日志集成到代理中（可选）：

# in agents/specialists.py after receiving content
from infra.obs import log_event
# ... inside chat(...)
resp = r.json()
log_event("chat_request", {"endpoint": f"{BASE_URL}/v1/chat/completions"})
log_event("chat_response", resp)
return resp["choices"][0]["message"]["content"]
​

步骤 2）通过 CLI 验证可用性和基本健康状况

REM Ensure Foundry Local is running a model
foundry model list
foundry model run phi-4-mini

REM Validate the OpenAI-compatible endpoint
curl http://localhost:8000/v1/models
​

步骤 3）数据清理和 PII 卫生

• 在将消息发送到模型之前，剔除或哈希敏感字段（如电子邮件、电话号码、ID）
• 将原始源数据保留在设备上，仅传递必要的上下文字符串

示例清理辅助工具：

# infra/redact.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
​

在代理中使用：

from infra.redact import sanitize
# user_goal = sanitize(user_goal)
# context = sanitize(context)
​

步骤 4）断路器和错误处理

• 使用 try/except 和指数回退包装每个代理调用
• 在重复失败时短路管道

import time

def with_retry(func, retries=3, base_delay=0.5):
    for i in range(retries):
        try:
            return func()
        except Exception as e:
            if i == retries - 1:
                raise
            time.sleep(base_delay * (2 ** i))
​

步骤 5）本地审计日志和导出

• 将 JSON 日志存储在 ./agent_logs 下
• 定期压缩和轮换日志
• 导出摘要以供审查（计数、平均延迟、错误率）

步骤 6）与 Microsoft Learn 文档交叉检查

• Foundry Local 提供 OpenAI 兼容 API（通过 curl /v1/models 验证）
• 使用 foundry model run <name> 确认模型可用性
• 遵循官方指导进行客户端集成和示例应用（Open WebUI/操作指南）

参考资料：

• Foundry Local 文档：https://learn.microsoft.com/en-us/azure/ai-foundry/foundry-local/
• Azure AI Agents：https://learn.microsoft.com/en-us/azure/ai-services/agents/overview
• 本地示例：
  • 多代理编排：Module08/samples/05/multi_agent_orchestration.ipynb
  • 代理实现：Module08/samples/05/agents/
  • 示例 README：Module08/samples/05/README.md
• 官方 Microsoft 示例：
  • 函数调用
  • Hello Foundry Local
  • Foundry Local Python SDK
• 集成示例：https://learn.microsoft.com/en-us/azure/ai-foundry/foundry-local/how-to/how-to-chat-application-with-open-web-ui

下一步

• 探索 Azure AI Agents 用于云托管编排
• 添加企业连接器（Microsoft Graph、搜索、数据库）