{
"cells": [
{
"cell_type": "markdown",
"id": "192a5723",
"metadata": {},
"source": [
"# 第17课 - 使用 Foundry Local 和 Qwen 创建本地 AI 代理\n",
"\n",
"在本笔记本中，您将构建一个本地工程助理，它完全运行在您的工作站上。整个过程中不使用任何云推理。该助理可以：\n",
"\n",
"1. 通过 Foundry Local 的 Qwen 函数调用调用工具。\n",
"2. 列出和读取沙盒项目目录中的文件。\n",
"3. 用简单指标分析代码。\n",
"4. 通过本地 RAG（Chroma）搜索文档。\n",
"5. **使用本地 MCP 服务器**（如果未配置则优雅跳过）。\n",
"\n",
"代理代码几乎与云端课程相同——唯一不同的是客户端端点从云端移到了`localhost`。\n"
]
},
{
"cell_type": "markdown",
"id": "40f2192b",
"metadata": {},
"source": [
"## 设置\n",
"\n",
"在运行此笔记本之前：\n",
"\n",
"1. **安装 Microsoft Foundry Local**（请参阅适用于您的操作系统的[文档](https://learn.microsoft.com/azure/ai-foundry/foundry-local/)）。\n",
"2. **下载并启动 Qwen 模型：**\n",
" ```bash\n",
" foundry model run qwen2.5-7b-instruct\n",
" foundry service status\n",
" ```\n",
"3. 安装以下 Python 包。\n",
"\n",
"所有操作均在本地运行。配备约 8 GB 内存的机器是一个现实的最低要求。\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "80565bf0",
"metadata": {},
"outputs": [],
"source": [
"%pip install foundry-local-sdk openai chromadb -q"
]
},
{
"cell_type": "markdown",
"id": "f815e0c2",
"metadata": {},
"source": [
"## 1. 连接到 Foundry Local\n",
"\n",
"`FoundryLocalManager` 会在需要时下载模型，启动本地服务，并为我们提供一个 **兼容 OpenAI 的端点**。然后我们将标准的 OpenAI SDK 指向该端点。API 密钥是本地占位符 — 不涉及任何云端凭证。\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "248ef257",
"metadata": {},
"outputs": [],
"source": [
"from foundry_local import FoundryLocalManager\n",
"from openai import OpenAI\n",
"\n",
"MODEL_ALIAS = \"qwen2.5-7b-instruct\"\n",
"\n",
"# Foundry Local selects the best build for your hardware (CPU / GPU / NPU) automatically.\n",
"manager = FoundryLocalManager(MODEL_ALIAS)\n",
"model_info = manager.get_model_info(MODEL_ALIAS)\n",
"\n",
"client = OpenAI(\n",
" base_url=manager.endpoint, # e.g. http://localhost:PORT/v1\n",
" api_key=manager.api_key, # local placeholder\n",
")\n",
"\n",
"MODEL_ID = model_info.id\n",
"print(f\"Connected to Foundry Local. Serving: {MODEL_ID}\")\n",
"print(f\"Endpoint: {manager.endpoint}\")"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "9ff96f0b",
"metadata": {},
"outputs": [],
"source": [
"# Quick sanity check: a plain chat completion, running fully on-device.\n",
"resp = client.chat.completions.create(\n",
" model=MODEL_ID,\n",
" messages=[{\"role\": \"user\", \"content\": \"In one sentence, what is a small language model?\"}],\n",
")\n",
"print(resp.choices[0].message.content)"
]
},
{
"cell_type": "markdown",
"id": "1f01c3be",
"metadata": {},
"source": [
"## 2. 本地工具（沙盒文件操作）\n",
"\n",
"我们在磁盘上创建一个小型示例项目，然后定义作用于该项目根目录的工具。即使在本地，沙盒检查也很重要：一个读取任意路径的工具运行时权限与您的用户权限相同，可以访问您能访问的任何内容。\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "c7c472f3",
"metadata": {},
"outputs": [],
"source": [
"import json\n",
"from pathlib import Path\n",
"\n",
"# Create a tiny sample project so the notebook is self-contained.\n",
"PROJECT_ROOT = Path.cwd() / \"sample_project\"\n",
"PROJECT_ROOT.mkdir(exist_ok=True)\n",
"\n",
"(PROJECT_ROOT / \"auth.py\").write_text(\n",
" '\"\"\"Authentication helpers.\"\"\"\\n\\n'\n",
" \"def login(user, password):\\n\"\n",
" \" # TODO: hash the password before comparing\\n\"\n",
" \" return user == 'admin' and password == 'secret'\\n\\n\"\n",
" \"def logout(session):\\n\"\n",
" \" session.clear()\\n\",\n",
" encoding=\"utf-8\",\n",
")\n",
"(PROJECT_ROOT / \"utils.py\").write_text(\n",
" '\"\"\"Utility functions.\"\"\"\\n\\n'\n",
" \"def clamp(value, low, high):\\n\"\n",
" \" return max(low, min(value, high))\\n\",\n",
" encoding=\"utf-8\",\n",
")\n",
"print(\"Sample project created at:\", PROJECT_ROOT)"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "9cf382db",
"metadata": {},
"outputs": [],
"source": [
"def _safe_path(path: str) -> Path | None:\n",
" \"\"\"Resolve a path and confirm it stays inside the project sandbox.\"\"\"\n",
" full = (PROJECT_ROOT / path).resolve()\n",
" if full == PROJECT_ROOT or PROJECT_ROOT in full.parents:\n",
" return full\n",
" return None\n",
"\n",
"\n",
"def list_files() -> str:\n",
" \"\"\"List files in the project directory.\"\"\"\n",
" files = [p.name for p in PROJECT_ROOT.iterdir() if p.is_file()]\n",
" return \", \".join(files) if files else \"(no files)\"\n",
"\n",
"\n",
"def read_file(path: str) -> str:\n",
" \"\"\"Read a file, but only inside the sandboxed project directory.\"\"\"\n",
" full = _safe_path(path)\n",
" if full is None:\n",
" return \"Access denied: path is outside the project directory.\"\n",
" if not full.is_file():\n",
" return f\"No such file: {path}\"\n",
" return full.read_text(encoding=\"utf-8\")\n",
"\n",
"\n",
"def analyze_code(path: str) -> str:\n",
" \"\"\"Report simple metrics about a source file.\"\"\"\n",
" full = _safe_path(path)\n",
" if full is None or not full.is_file():\n",
" return \"File not found or access denied.\"\n",
" text = full.read_text(encoding=\"utf-8\")\n",
" lines = text.splitlines()\n",
" return json.dumps({\n",
" \"path\": path,\n",
" \"lines\": len(lines),\n",
" \"functions\": sum(1 for ln in lines if ln.strip().startswith(\"def \")),\n",
" \"todos\": sum(1 for ln in lines if \"TODO\" in ln or \"FIXME\" in ln),\n",
" })\n",
"\n",
"\n",
"print(list_files())"
]
},
{
"cell_type": "markdown",
"id": "49489826",
"metadata": {},
"source": [
"## 3. 使用 Chroma 的本地 RAG\n",
"\n",
"我们将一小部分文档片段嵌入到本地 Chroma 集合中。Chroma 在进程内运行并将向量存储在磁盘上——无需服务器，无需云。`search_docs` 工具检索与查询最相关的片段。\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "43f9333e",
"metadata": {},
"outputs": [],
"source": [
"import chromadb\n",
"\n",
"DOCS = {\n",
" \"auth\": \"The login() function checks credentials. It currently compares passwords in plain text, which should be hashed.\",\n",
" \"sessions\": \"Sessions are cleared on logout via session.clear(). Sessions are stored in memory and lost on restart.\",\n",
" \"utils\": \"clamp(value, low, high) constrains a number to a range. Used throughout the UI layer for bounds checking.\",\n",
" \"style\": \"This project follows PEP 8. Functions use snake_case and modules include a docstring at the top.\",\n",
"}\n",
"\n",
"# Chroma ships with a local default embedding model, so embedding stays on-device.\n",
"chroma_client = chromadb.Client()\n",
"collection = chroma_client.get_or_create_collection(\"project_docs\")\n",
"collection.upsert(\n",
" ids=list(DOCS.keys()),\n",
" documents=list(DOCS.values()),\n",
")\n",
"\n",
"\n",
"def search_docs(query: str) -> str:\n",
" \"\"\"Search the local documentation index for relevant snippets.\"\"\"\n",
" results = collection.query(query_texts=[query], n_results=2)\n",
" docs = results.get(\"documents\", [[]])[0]\n",
" return \"\\n\".join(docs) if docs else \"No relevant documentation found.\"\n",
"\n",
"\n",
"print(search_docs(\"how are passwords handled?\"))"
]
},
{
"cell_type": "markdown",
"id": "7d7ed86c",
"metadata": {},
"source": [
"## 4. 工具调用循环\n",
"\n",
"现在我们使用 OpenAI 工具架构向模型注册工具，并运行标准的工具调用循环——模型请求一个工具，我们在本地执行该工具，将结果反馈回去，并重复该过程，直到模型生成最终答案。Qwen 的可靠函数调用使得这能在设备上运行。\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "8b7f06cd",
"metadata": {},
"outputs": [],
"source": [
"TOOLS_SCHEMA = [\n",
" {\"type\": \"function\", \"function\": {\n",
" \"name\": \"list_files\", \"description\": \"List files in the project directory.\",\n",
" \"parameters\": {\"type\": \"object\", \"properties\": {}},\n",
" }},\n",
" {\"type\": \"function\", \"function\": {\n",
" \"name\": \"read_file\", \"description\": \"Read a file inside the project directory.\",\n",
" \"parameters\": {\"type\": \"object\", \"properties\": {\n",
" \"path\": {\"type\": \"string\", \"description\": \"File name, e.g. auth.py\"}}, \"required\": [\"path\"]},\n",
" }},\n",
" {\"type\": \"function\", \"function\": {\n",
" \"name\": \"analyze_code\", \"description\": \"Report line count, function count and TODO count for a file.\",\n",
" \"parameters\": {\"type\": \"object\", \"properties\": {\n",
" \"path\": {\"type\": \"string\"}}, \"required\": [\"path\"]},\n",
" }},\n",
" {\"type\": \"function\", \"function\": {\n",
" \"name\": \"search_docs\", \"description\": \"Search local documentation for a query.\",\n",
" \"parameters\": {\"type\": \"object\", \"properties\": {\n",
" \"query\": {\"type\": \"string\"}}, \"required\": [\"query\"]},\n",
" }},\n",
"]\n",
"\n",
"TOOL_IMPL = {\n",
" \"list_files\": list_files,\n",
" \"read_file\": read_file,\n",
" \"analyze_code\": analyze_code,\n",
" \"search_docs\": search_docs,\n",
"}\n",
"\n",
"SYSTEM_PROMPT = (\n",
" \"You are a local engineering assistant. Use the provided tools to inspect the project \"\n",
" \"and its documentation. Prefer calling a tool over guessing. Be concise.\"\n",
")"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "dbb0778c",
"metadata": {},
"outputs": [],
"source": [
"def run_agent(user_query: str, max_iterations: int = 5) -> str:\n",
" \"\"\"Standard tool-calling loop, running entirely against the local model.\"\"\"\n",
" messages = [\n",
" {\"role\": \"system\", \"content\": SYSTEM_PROMPT},\n",
" {\"role\": \"user\", \"content\": user_query},\n",
" ]\n",
"\n",
" for _ in range(max_iterations):\n",
" response = client.chat.completions.create(\n",
" model=MODEL_ID,\n",
" messages=messages,\n",
" tools=TOOLS_SCHEMA,\n",
" )\n",
" msg = response.choices[0].message\n",
"\n",
" if not msg.tool_calls:\n",
" return msg.content or \"(no answer)\"\n",
"\n",
" # Record the assistant's tool-call request.\n",
" messages.append({\n",
" \"role\": \"assistant\",\n",
" \"content\": msg.content,\n",
" \"tool_calls\": [tc.model_dump() for tc in msg.tool_calls],\n",
" })\n",
"\n",
" # Execute each requested tool locally and feed results back.\n",
" for tc in msg.tool_calls:\n",
" name = tc.function.name\n",
" args = json.loads(tc.function.arguments or \"{}\")\n",
" result = TOOL_IMPL[name](**args) if name in TOOL_IMPL else f\"Unknown tool: {name}\"\n",
" messages.append({\n",
" \"role\": \"tool\",\n",
" \"tool_call_id\": tc.id,\n",
" \"content\": str(result),\n",
" })\n",
"\n",
" return \"Stopped: reached max tool-calling iterations.\"\n",
"\n",
"\n",
"print(\"Agent ready.\")"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "9f0c0c05",
"metadata": {},
"outputs": [],
"source": [
"# A file-reading question.\n",
"print(run_agent(\"What does auth.py do, and is there anything to fix in it?\"))"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "80646664",
"metadata": {},
"outputs": [],
"source": [
"# A RAG question.\n",
"print(run_agent(\"According to the docs, how are passwords currently handled?\"))"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "f321f218",
"metadata": {},
"outputs": [],
"source": [
"# A code-analysis question.\n",
"print(run_agent(\"How many functions and TODOs are in auth.py?\"))"
]
},
{
"cell_type": "markdown",
"id": "41cedd11",
"metadata": {},
"source": [
"## 5. 本地 MCP（可选）\n",
"\n",
"MCP 是一种传输方式，而非云服务——MCP 服务器可以作为一个本地进程通过 `stdio` 运行。下面的单元展示了如果你配置了一个本地 MCP 服务器（例如文件系统服务器），如何连接它。当 `LOCAL_MCP_COMMAND` 未设置时，它会优雅地跳过，因此笔记本仍能端到端运行。\n",
"\n",
"安全提示：本地 MCP 服务器以你的用户权限运行。请将其限定在项目目录范围内，并在使用其输出之前进行验证。\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "9250c63f",
"metadata": {},
"outputs": [],
"source": [
"import os\n",
"\n",
"LOCAL_MCP_COMMAND = os.getenv(\"LOCAL_MCP_COMMAND\") # e.g. \"npx -y @modelcontextprotocol/server-filesystem ./sample_project\"\n",
"\n",
"if not LOCAL_MCP_COMMAND:\n",
" print(\"No LOCAL_MCP_COMMAND set — skipping the MCP demo. \"\n",
" \"Set it to a local MCP server command to try this section.\")\n",
"else:\n",
" import asyncio\n",
" from mcp import ClientSession, StdioServerParameters\n",
" from mcp.client.stdio import stdio_client\n",
"\n",
" async def list_mcp_tools(command: str):\n",
" parts = command.split()\n",
" params = StdioServerParameters(command=parts[0], args=parts[1:])\n",
" async with stdio_client(params) as (read, write):\n",
" async with ClientSession(read, write) as session:\n",
" await session.initialize()\n",
" tools = await session.list_tools()\n",
" return [t.name for t in tools.tools]\n",
"\n",
" names = await list_mcp_tools(LOCAL_MCP_COMMAND)\n",
" print(\"Local MCP server exposes tools:\", names)"
]
},
{
"cell_type": "markdown",
"id": "e4828211",
"metadata": {},
"source": [
"## 摘要\n",
"\n",
"你构建了一个完全在你的机器上运行的工程助理：\n",
"\n",
"- **Foundry Local** 在一个兼容OpenAI的端点后面提供了一个 **Qwen** 模型——因此代理代码与云端课程相匹配。\n",
"- 沙盒工具 让代理获得文件访问和代码分析能力，而无需离开项目目录。\n",
"- **Chroma** 提供了针对文档的 **本地RAG**。\n",
"- **本地MCP** 展示了如何离线重用MCP生态系统。\n",
"\n",
"在任何阶段都没有使用云端推理。\n",
"\n",
"### 挑战\n",
"\n",
"扩展本地代理以：\n",
"\n",
"1. **支持多个MCP服务器** —— 连接一个文件系统服务器和一个git服务器，让代理能在它们之间做出选择。\n",
"2. 使用本地内存 —— 将简短的对话历史持久化到磁盘，以便助手能在笔记本重启后记住之前的对话内容。\n",
"3. 支持本地多代理编排 —— 添加第二个本地代理（例如审阅者），让两个代理协作完成任务。\n",
"\n",
"在下一课中，你将学习如何保护部署的AI代理。\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"---\n\n\n**免责声明**：\n本文件由 AI 翻译服务 [Co-op Translator](https://github.com/Azure/co-op-translator) 翻译完成。尽管我们力求准确，但请注意，自动翻译可能包含错误或不准确之处。原始语言版文件应视为权威来源。对于重要信息，建议使用专业人工翻译。我们对因使用本翻译而产生的任何误解或误释不承担责任。\n\n"
]
}
],
"metadata": {
"language_info": {
"name": "python"
}
},
"nbformat": 4,
"nbformat_minor": 5
}