在实际项目中将本地大模型、技能框架和智能体Agent能力结合起来构建一个能管理本地系统、处理复杂任务的自动化助手正成为一个越来越普遍的需求。很多开发者尝试使用 Ollama 部署本地模型却发现模型本身缺乏“行动”能力无法调用系统API、读取文件或执行命令而一些客户端工具虽然能管理模型却又没有提供灵活的智能体编排逻辑。这导致了一个常见的困境模型很“聪明”但“手”被束缚住了。本文旨在解决这个核心问题如何将DeepSeek R1 7B这类本地模型、LifeOS Skill这样的技能框架与一个具备执行能力的本地 Agent客户端有效整合形成一个既能理解复杂指令又能安全、可靠地操作本地系统的完整解决方案。我们将从概念梳理开始逐步搭建一个可运行的原型并重点解释其中的配置要点、通信机制和常见排查路径。无论你是想为个人工作流添加自动化还是探索本地AI应用开发这篇文章都将提供一个清晰的实践指南。1. 核心概念解析模型、技能与智能体的分工在开始动手之前必须厘清三个核心组件各自的角色和它们之间的协作关系。混淆概念是后续集成失败的主要原因。1.1 本地大模型负责“思考”与“规划”本地大模型例如通过 Ollama 运行的deepseek-r1:7b、qwen2.5:7b或hermes2-pro是整个系统的“大脑”。它的核心职责是自然语言理解和任务规划。输入用户的自然语言指令如“帮我总结上个月项目文档的修改要点”。处理模型将指令分解为一系列可执行的子步骤或“思维链”并判断需要调用哪些工具技能。输出结构化的响应通常是一个包含工具调用请求的 JSON 对象例如{action: read_file, params: {path: /docs/project_log.md}}。关键限制模型本身是一个“沙盒”它只能生成文本无法直接操作文件系统、网络或运行命令。它需要“执行器”。1.2 技能标准化的“工具”或“能力”技能是预先定义好的、可被调用的原子操作。LifeOS Skill或类似框架定义了一套技能的标准接口和描述方式。作用将复杂的系统操作如读写文件、调用API、执行Shell命令、查询数据库封装成统一的、模型可理解的“工具”。格式一个技能通常包含名称、描述、输入参数列表类型、说明和具体的执行函数。示例一个file_search技能描述为“在指定目录中搜索包含关键词的文件”参数为directory和keyword。当模型决定需要搜索文件时就会生成调用此技能的请求。1.3 本地 Agent负责“调度”与“执行”本地 Agent 是系统的“中枢神经系统”和“双手”。它负责协调模型和技能是真正让想法落地的一环。核心职责会话管理维护与用户的对话历史提供上下文给模型。工具调用接收模型输出的工具调用请求找到对应的技能并执行。安全控制在执行敏感操作如rm命令前进行确认或权限校验。结果反馈将技能执行的结果成功或失败重新组织成自然语言或作为新的上下文返回给模型进行下一步决策。常见形态可以是一个独立的 Python 脚本、一个桌面应用如某些 AI 助手客户端或一个后台服务。三者协作的典型流程如下用户向Agent发出指令。Agent将指令和对话历史发送给本地模型。模型思考后返回一个包含技能调用请求的响应。Agent解析响应加载对应的技能并执行。技能执行完毕将结果返回给Agent。Agent将结果反馈给模型模型根据结果决定下一步继续调用技能或生成最终回答。Agent将最终结果呈现给用户。2. 环境准备与组件选型基于输入材料中提到的热词我们构建一个以Ollama DeepSeek R1 7B 自定义 Python Agent 简易技能的技术栈。这个组合兼顾了性能、可控性和开发灵活性。2.1 基础运行环境操作系统Windows 10/11, macOS, 或 Linux。本文以 Windows 10 为例命令会有相应调整。Python3.8 或更高版本。这是编写 Agent 和技能的主要语言。包管理工具pip。建议使用虚拟环境venv或conda隔离依赖。2.2 核心组件安装与验证2.2.1 Ollama 部署与模型拉取Ollama 是运行和管理本地大模型的利器。安装 Ollama访问 Ollama 官网下载对应操作系统的安装包并安装。安装完成后打开终端Windows 为 PowerShell 或 CMD运行ollama --version验证安装。拉取 DeepSeek R1 7B 模型ollama pull deepseek-r1:7b这个命令会从仓库下载模型耗时取决于网络。deepseek-r1:7b是一个经过强化学习训练、擅长推理和规划的模型非常适合 Agent 场景。你也可以选择其他模型如qwen2.5:7b、llama3.1:8b或hermes2-pro只需替换名称即可。验证模型运行ollama run deepseek-r1:7b成功运行后会进入一个交互式对话界面输入“你好”测试。输入/bye退出。2.2.2 构建本地 Agent 与技能框架我们将使用 Python 的requests库与 Ollama 的 API 通信并构建一个简单的技能框架。首先创建项目目录结构local_ai_agent/ ├── agent_core.py # Agent 核心调度逻辑 ├── skills/ # 技能包目录 │ ├── __init__.py │ ├── file_ops.py # 文件操作技能 │ └── system_info.py # 系统信息技能 ├── config.yaml # 配置文件 ├── requirements.txt # 依赖列表 └── main.py # 程序入口创建虚拟环境并安装依赖# 在项目根目录下 python -m venv venv # Windows 激活 venv\Scripts\activate # macOS/Linux 激活 # source venv/bin/activate # 编辑 requirements.txt内容如下 # requests2.28.0 # pyyaml6.0 # python-dotenv1.0.0 pip install -r requirements.txt编写配置文件config.yamlollama: base_url: http://localhost:11434 # Ollama 默认 API 地址 model: deepseek-r1:7b # 使用的模型名称 temperature: 0.1 # 较低的温度使输出更确定适合工具调用 agent: name: LocalDeepSeekAgent max_iterations: 10 # 防止 Agent 陷入死循环 skill_timeout: 30 # 技能执行超时时间秒 skills: enabled: - file_read - file_write - get_current_time - list_directory # 可以在这里为技能配置参数如允许访问的根目录 file_root: C:\\Users\\YourName\\AgentWorkspace # Windows 路径示例3. 实现简易技能与 Agent 核心3.1 实现技能技能是实现具体功能的地方。每个技能都是一个 Python 函数并附带元数据描述。skills/file_ops.py:import os import json from datetime import datetime from pathlib import Path from typing import Dict, Any def read_file(params: Dict[str, Any]) - Dict[str, Any]: 读取指定文件的内容。 参数: {“path”: “文件路径字符串”} try: file_path Path(params.get(path)) # 简单的安全限制检查路径是否在允许的根目录下实际项目需更严格 # allowed_root Path(config.get(skills, {}).get(file_root, .)) # if not file_path.resolve().is_relative_to(allowed_root.resolve()): # return {success: False, error: Access denied: path out of allowed scope.} if not file_path.is_file(): return {success: False, error: fFile not found: {file_path}} with open(file_path, r, encodingutf-8) as f: content f.read() return {success: True, content: content} except Exception as e: return {success: False, error: str(e)} def write_file(params: Dict[str, Any]) - Dict[str, Any]: 将内容写入指定文件覆盖。 参数: {“path”: “文件路径”, “content”: “要写入的文本内容”} try: file_path Path(params.get(path)) content params.get(content, ) # 安全限制同上 file_path.parent.mkdir(parentsTrue, exist_okTrue) with open(file_path, w, encodingutf-8) as f: f.write(content) return {success: True, message: fFile written: {file_path}} except Exception as e: return {success: False, error: str(e)} def list_directory(params: Dict[str, Any]) - Dict[str, Any]: 列出指定目录下的文件和文件夹。 参数: {“path”: “目录路径” 可选默认为当前目录} try: target_path Path(params.get(path, .)) if not target_path.is_dir(): return {success: False, error: fNot a directory: {target_path}} items [] for item in target_path.iterdir(): items.append({ name: item.name, type: directory if item.is_dir() else file, size: item.stat().st_size if item.is_file() else 0 }) return {success: True, items: items} except Exception as e: return {success: False, error: str(e)} # 技能元数据用于告知模型此技能的功能和调用方式 FILE_OPS_SKILLS { read_file: { function: read_file, description: 读取一个文本文件的内容。, parameters: { path: {type: string, description: 要读取的文件的完整路径。} } }, write_file: { function: write_file, description: 将文本内容写入一个文件。如果文件不存在则创建存在则覆盖。, parameters: { path: {type: string, description: 要写入的文件的完整路径。}, content: {type: string, description: 要写入的文本内容。} } }, list_directory: { function: list_directory, description: 列出指定目录下的所有文件和子目录。, parameters: { path: {type: string, description: 要列出的目录路径。默认为当前目录。, required: False} } } }skills/system_info.py:from datetime import datetime import platform from typing import Dict, Any def get_current_time(params: Dict[str, Any]) - Dict[str, Any]: 获取当前系统时间。无需参数。 try: current_time datetime.now().strftime(%Y-%m-%d %H:%M:%S) return {success: True, current_time: current_time} except Exception as e: return {success: False, error: str(e)} def get_system_info(params: Dict[str, Any]) - Dict[str, Any]: 获取基本的系统信息。无需参数。 try: info { system: platform.system(), release: platform.release(), machine: platform.machine(), processor: platform.processor(), } return {success: True, info: info} except Exception as e: return {success: False, error: str(e)} SYSTEM_INFO_SKILLS { get_current_time: { function: get_current_time, description: 获取当前的日期和时间。, parameters: {} }, get_system_info: { function: get_system_info, description: 获取操作系统和硬件的基本信息。, parameters: {} } }3.2 实现 Agent 核心调度逻辑Agent 的核心是循环问模型 - 解析工具调用 - 执行技能 - 反馈结果 - 继续问模型。agent_core.py:import requests import json import yaml from typing import Dict, List, Any, Optional import time from skills.file_ops import FILE_OPS_SKILLS from skills.system_info import SYSTEM_INFO_SKILLS class LocalAIAgent: def __init__(self, config_path: str config.yaml): with open(config_path, r, encodingutf-8) as f: self.config yaml.safe_load(f) self.ollama_url self.config[ollama][base_url] self.model self.config[ollama][model] self.max_iterations self.config[agent][max_iterations] # 加载启用的技能 self.skills {} self._load_skills() # 对话历史 self.conversation_history [] def _load_skills(self): 根据配置加载技能。 enabled_skills self.config[skills][enabled] all_skill_dicts [FILE_OPS_SKILLS, SYSTEM_INFO_SKILLS] for skill_dict in all_skill_dicts: for skill_name, skill_info in skill_dict.items(): if skill_name in enabled_skills: self.skills[skill_name] skill_info print(fLoaded skills: {list(self.skills.keys())}) def _call_ollama(self, messages: List[Dict]) - Dict[str, Any]: 调用 Ollama API 进行聊天补全。 payload { model: self.model, messages: messages, stream: False, options: { temperature: self.config[ollama].get(temperature, 0.1) } } try: response requests.post( f{self.ollama_url}/api/chat, jsonpayload, timeout60 ) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: raise Exception(fOllama API call failed: {e}) def _extract_tool_call(self, response_message: Dict) - Optional[Dict]: 从模型回复中解析工具调用请求。 期望模型回复的 content 字段是一个 JSON 字符串格式如 {action: skill_name, params: {...}} 或者是一个包含类似结构的思考文本。 这是一个简化的解析器实际应用可能需要更复杂的逻辑或使用 OpenAI 兼容的 tool_calls 格式。 content response_message.get(content, ).strip() # 尝试从 JSON 块中解析 import re json_match re.search(r\{.*action.*\}, content, re.DOTALL) if json_match: try: tool_call json.loads(json_match.group()) if action in tool_call and params in tool_call: return tool_call except json.JSONDecodeError: pass # 如果模型直接给出了最终答案则返回 None return None def _execute_skill(self, skill_name: str, params: Dict) - Dict[str, Any]: 执行指定的技能。 if skill_name not in self.skills: return {success: False, error: fSkill {skill_name} not found or disabled.} skill_info self.skills[skill_name] skill_func skill_info[function] try: # 在实际项目中这里应加入更严格的参数验证和权限检查 result skill_func(params) return result except Exception as e: return {success: False, error: fSkill execution error: {str(e)}} def run(self, user_input: str) - str: 运行 Agent 处理用户输入。 print(f\n[User]: {user_input}) # 初始化消息历史可以加入系统提示词来引导模型使用工具 system_prompt f你是一个有帮助的AI助手可以调用工具来帮助用户。你可以使用的工具如下 {json.dumps([{name: k, description: v[description], parameters: v[parameters]} for k, v in self.skills.items()], indent2)} 当你需要调用工具时请严格按照以下JSON格式回复且只回复这个JSON对象不要添加其他解释 {{action: 工具名称, params: {{参数名: 参数值}}}} 如果用户的问题不需要调用工具或者工具调用完成后得到了最终答案请直接以自然语言回复。 messages [ {role: system, content: system_prompt}, *self.conversation_history[-6:], # 保留最近几轮历史作为上下文 {role: user, content: user_input} ] iteration 0 final_answer None while iteration self.max_iterations: iteration 1 print(f\n[Agent] Iteration {iteration}: Querying model...) # 1. 调用模型 ollama_response self._call_ollama(messages) assistant_message ollama_response[message] messages.append(assistant_message) # 将模型回复加入历史 # 2. 尝试解析工具调用 tool_call self._extract_tool_call(assistant_message) if tool_call: action tool_call[action] params tool_call[params] print(f[Agent] Detected tool call: {action} with params {params}) # 3. 执行工具 tool_result self._execute_skill(action, params) print(f[Agent] Tool result: {tool_result}) # 4. 将工具执行结果作为新的上下文反馈给模型 result_message { role: user, content: fThe result of tool {action} is: {json.dumps(tool_result)}. Please continue based on this result. } messages.append(result_message) # 继续下一轮循环让模型基于结果决定下一步 else: # 模型没有调用工具直接给出了最终答案 final_answer assistant_message[content] print(f[Agent] Model provided final answer.) break if not final_answer and iteration self.max_iterations: final_answer 已达到最大迭代次数未能完成请求。 # 将本轮完整交互加入历史可控制历史长度 self.conversation_history.extend([ {role: user, content: user_input}, {role: assistant, content: final_answer or No answer generated.} ]) return final_answer or Agent finished without a final answer.4. 运行验证与结果分析4.1 创建程序入口并运行main.py:from agent_core import LocalAIAgent def main(): agent LocalAIAgent() print(Local AI Agent 已启动。输入 quit 或 exit 退出。) while True: try: user_input input(\n ) if user_input.lower() in [quit, exit, q]: print(再见) break if not user_input.strip(): continue answer agent.run(user_input) print(f\n[Assistant]: {answer}) except KeyboardInterrupt: print(\n程序被中断。) break except Exception as e: print(f\n[Error]: 处理请求时出错 - {e}) if __name__ __main__: main()4.2 启动与测试确保 Ollama 服务运行在另一个终端执行ollama serve或确保 Ollama 后台服务已启动。运行 Agent在项目根目录下激活虚拟环境后运行python main.py看到Loaded skills: [read_file, write_file, get_current_time, list_directory]和提示符即表示成功。进行功能测试测试系统技能 现在几点了 [Agent] Iteration 1: Querying model... [Agent] Detected tool call: get_current_time with params {} [Agent] Tool result: {success: True, current_time: 2024-01-01 15:30:22} [Agent] Iteration 2: Querying model... [Agent] Model provided final answer. [Assistant]: 当前时间是 2024-01-01 15:30:22。测试文件操作技能 请在我的文档文件夹假设是 C:\Users\YourName\AgentWorkspace里创建一个名为 test_agent.txt 的文件内容写“Hello from DeepSeek Agent”。观察 Agent 的思考过程它应该会调用write_file技能。执行成功后你可以去对应目录查看文件。测试复杂任务 请列出我的文档文件夹里有什么文件然后告诉我文件数量。Agent 会先调用list_directory获取结果后模型会理解结果并计算数量最终给出回答。4.3 结果分析成功标志Agent 能正确理解你的意图将任务分解调用正确的技能并将执行结果整合成自然语言回复。模型规划能力deepseek-r1:7b这类模型在收到清晰的系统提示和工具描述后通常能较好地规划简单任务。对于多步骤任务需要依赖其推理能力。系统健壮性当前实现是一个基础原型。生产环境需要更完善的错误处理、技能执行超时控制、对话历史管理以及防止技能被滥用的安全沙箱。5. 常见问题排查与调试在整合过程中你可能会遇到以下典型问题。这里提供排查思路。问题现象可能原因检查方式处理建议启动 Agent 时报连接 Ollama 失败1. Ollama 服务未启动。2.config.yaml中的base_url端口错误。3. 防火墙或网络策略阻止连接。1. 在浏览器访问http://localhost:11434看是否返回 Ollama 信息。2. 运行ollama list命令确认服务正常。3. 检查config.yaml配置。1. 确保先运行ollama serve。2. 确认 Ollama 版本API 端口是否为默认的 11434。3. 暂时关闭防火墙或检查代理设置。模型不调用工具总是直接回答1. 系统提示词system_prompt不够清晰或格式不对。2. 模型能力有限不理解工具调用格式。3. 技能描述不够准确。1. 打印出发送给模型的完整messages检查system_prompt。2. 尝试用更简单的指令测试如“使用 get_current_time 工具”。3. 换用qwen2.5:7b或llama3.1等工具调用能力更强的模型测试。1. 优化提示词明确要求模型以指定 JSON 格式回复。2. 在提示词中提供 1-2 个清晰的调用示例。3. 考虑使用支持 OpenAI 兼容tool_calls格式的模型和前端如 LiteLLM。技能执行失败返回权限错误1. 程序运行用户没有目标文件/目录的读写权限。2. 路径不存在或拼写错误。3. Windows 路径中的反斜杠未转义。1. 检查 Agent 进程的用户权限。2. 在技能函数内打印传入的params确认路径正确。3. 使用Path库处理路径它兼容不同操作系统。1. 为 Agent 设置一个安全的工作根目录并在技能中做路径解析和限制。2. 在技能函数开头添加详细的日志和参数验证。3. 使用绝对路径并确保路径存在。Agent 陷入无限循环1. 模型反复调用同一个工具或无效工具。2.max_iterations设置过高或逻辑有误。3. 工具执行结果格式混乱模型无法理解。1. 查看每次迭代中模型输出的content和解析出的tool_call。2. 检查_extract_tool_call函数是否能正确处理非工具调用回复。1. 降低max_iterations到 5-8。2. 在_extract_tool_call中加强判断如果content看起来是自然语言答案则停止循环。3. 确保工具返回的结果是结构化的字典。处理中文时出现乱码1. 文件读写编码不是utf-8。2. 控制台或终端编码不支持 UTF-8。1. 检查技能中open函数是否指定encodingutf-8。2. 在 Python 文件开头添加# -*- coding: utf-8 -*-。3. 检查终端编码Windows CMD 需使用chcp 65001。1. 统一使用 UTF-8 编码。2. 在 Windows 上使用 PowerShell 或支持 UTF-8 的终端如 Windows Terminal。3. 对输出字符串进行必要的编码处理。6. 进阶优化与生产环境建议上述原型验证了基本流程。要将其用于更严肃的场景需要考虑以下优化方向。6.1 技能框架标准化采用成熟框架考虑使用LangChain Tools、AutoGen的UserProxyAgent或CrewAI的Tool类来定义技能。它们提供了更标准的接口、更安全的执行环境和更丰富的工具库。技能动态注册与发现实现一个技能管理器支持热加载技能模块无需重启 Agent。6.2 提示工程优化结构化输出使用 Ollama 的format参数或模型本身的微调能力强制其以 JSON 等结构化格式输出使_extract_tool_call的解析更稳定。少样本示例在system_prompt中提供 2-3 个完整的“用户问题 - 模型思考 - 工具调用 - 结果 - 最终回答”的示例大幅提升模型遵循格式的能力。6.3 安全与权限控制至关重要技能沙箱对于执行 Shell 命令 (subprocess) 或高风险操作的技能必须在沙箱环境中运行限制其可访问的资源。用户确认在执行删除文件、修改系统设置等危险操作前Agent 应主动向用户请求确认。访问控制列表为不同技能配置 ACL定义其允许访问的文件路径、网络地址和系统调用。6.4 客户端与用户体验图形界面使用PyQt、Tkinter或web技术如Gradio、Streamlit构建一个简单的 GUI方便非技术用户使用。会话持久化将conversation_history保存到数据库或文件支持多轮对话和历史回顾。流式响应对接 Ollama 的流式 API实现类似 ChatGPT 的打字机效果提升体验。6.5 解决“上网查询信息受限”问题输入材料中提到“上网查询信息经常受限”这通常是因为技能缺失没有实现一个可以发起 HTTP 请求、解析网页内容的“网络搜索”技能。网络环境本地环境可能无法直接访问外部网络或需要配置代理。解决方案实现网络技能在skills/下创建web_search.py使用requests或aiohttp库封装搜索 API如 Serper、Google Custom Search或简单的网页抓取功能。务必遵守目标网站的robots.txt和服务条款。配置代理在技能函数或全局requests会话中配置代理设置以适配公司或地区的网络环境。使用本地知识库对于内部信息查询可以集成ChromaDB、FAISS等向量数据库将本地文档嵌入后供模型检索这是更安全、可控的方案。通过以上步骤你不仅能够搭建一个可运行的本地 AI Agent 系统还能理解其内部协作机制并具备排查问题和进行生产级优化的能力。核心在于明确模型、技能、Agent 三者的边界并通过稳定的通信协议如结构化 JSON将它们连接起来。 基于Ollama与DeepSeek R1 7B构建本地AI智能体：模型、技能与Agent的整合实践
在实际项目中将本地大模型、技能框架和智能体Agent能力结合起来构建一个能管理本地系统、处理复杂任务的自动化助手正成为一个越来越普遍的需求。很多开发者尝试使用 Ollama 部署本地模型却发现模型本身缺乏“行动”能力无法调用系统API、读取文件或执行命令而一些客户端工具虽然能管理模型却又没有提供灵活的智能体编排逻辑。这导致了一个常见的困境模型很“聪明”但“手”被束缚住了。本文旨在解决这个核心问题如何将DeepSeek R1 7B这类本地模型、LifeOS Skill这样的技能框架与一个具备执行能力的本地 Agent客户端有效整合形成一个既能理解复杂指令又能安全、可靠地操作本地系统的完整解决方案。我们将从概念梳理开始逐步搭建一个可运行的原型并重点解释其中的配置要点、通信机制和常见排查路径。无论你是想为个人工作流添加自动化还是探索本地AI应用开发这篇文章都将提供一个清晰的实践指南。1. 核心概念解析模型、技能与智能体的分工在开始动手之前必须厘清三个核心组件各自的角色和它们之间的协作关系。混淆概念是后续集成失败的主要原因。1.1 本地大模型负责“思考”与“规划”本地大模型例如通过 Ollama 运行的deepseek-r1:7b、qwen2.5:7b或hermes2-pro是整个系统的“大脑”。它的核心职责是自然语言理解和任务规划。输入用户的自然语言指令如“帮我总结上个月项目文档的修改要点”。处理模型将指令分解为一系列可执行的子步骤或“思维链”并判断需要调用哪些工具技能。输出结构化的响应通常是一个包含工具调用请求的 JSON 对象例如{action: read_file, params: {path: /docs/project_log.md}}。关键限制模型本身是一个“沙盒”它只能生成文本无法直接操作文件系统、网络或运行命令。它需要“执行器”。1.2 技能标准化的“工具”或“能力”技能是预先定义好的、可被调用的原子操作。LifeOS Skill或类似框架定义了一套技能的标准接口和描述方式。作用将复杂的系统操作如读写文件、调用API、执行Shell命令、查询数据库封装成统一的、模型可理解的“工具”。格式一个技能通常包含名称、描述、输入参数列表类型、说明和具体的执行函数。示例一个file_search技能描述为“在指定目录中搜索包含关键词的文件”参数为directory和keyword。当模型决定需要搜索文件时就会生成调用此技能的请求。1.3 本地 Agent负责“调度”与“执行”本地 Agent 是系统的“中枢神经系统”和“双手”。它负责协调模型和技能是真正让想法落地的一环。核心职责会话管理维护与用户的对话历史提供上下文给模型。工具调用接收模型输出的工具调用请求找到对应的技能并执行。安全控制在执行敏感操作如rm命令前进行确认或权限校验。结果反馈将技能执行的结果成功或失败重新组织成自然语言或作为新的上下文返回给模型进行下一步决策。常见形态可以是一个独立的 Python 脚本、一个桌面应用如某些 AI 助手客户端或一个后台服务。三者协作的典型流程如下用户向Agent发出指令。Agent将指令和对话历史发送给本地模型。模型思考后返回一个包含技能调用请求的响应。Agent解析响应加载对应的技能并执行。技能执行完毕将结果返回给Agent。Agent将结果反馈给模型模型根据结果决定下一步继续调用技能或生成最终回答。Agent将最终结果呈现给用户。2. 环境准备与组件选型基于输入材料中提到的热词我们构建一个以Ollama DeepSeek R1 7B 自定义 Python Agent 简易技能的技术栈。这个组合兼顾了性能、可控性和开发灵活性。2.1 基础运行环境操作系统Windows 10/11, macOS, 或 Linux。本文以 Windows 10 为例命令会有相应调整。Python3.8 或更高版本。这是编写 Agent 和技能的主要语言。包管理工具pip。建议使用虚拟环境venv或conda隔离依赖。2.2 核心组件安装与验证2.2.1 Ollama 部署与模型拉取Ollama 是运行和管理本地大模型的利器。安装 Ollama访问 Ollama 官网下载对应操作系统的安装包并安装。安装完成后打开终端Windows 为 PowerShell 或 CMD运行ollama --version验证安装。拉取 DeepSeek R1 7B 模型ollama pull deepseek-r1:7b这个命令会从仓库下载模型耗时取决于网络。deepseek-r1:7b是一个经过强化学习训练、擅长推理和规划的模型非常适合 Agent 场景。你也可以选择其他模型如qwen2.5:7b、llama3.1:8b或hermes2-pro只需替换名称即可。验证模型运行ollama run deepseek-r1:7b成功运行后会进入一个交互式对话界面输入“你好”测试。输入/bye退出。2.2.2 构建本地 Agent 与技能框架我们将使用 Python 的requests库与 Ollama 的 API 通信并构建一个简单的技能框架。首先创建项目目录结构local_ai_agent/ ├── agent_core.py # Agent 核心调度逻辑 ├── skills/ # 技能包目录 │ ├── __init__.py │ ├── file_ops.py # 文件操作技能 │ └── system_info.py # 系统信息技能 ├── config.yaml # 配置文件 ├── requirements.txt # 依赖列表 └── main.py # 程序入口创建虚拟环境并安装依赖# 在项目根目录下 python -m venv venv # Windows 激活 venv\Scripts\activate # macOS/Linux 激活 # source venv/bin/activate # 编辑 requirements.txt内容如下 # requests2.28.0 # pyyaml6.0 # python-dotenv1.0.0 pip install -r requirements.txt编写配置文件config.yamlollama: base_url: http://localhost:11434 # Ollama 默认 API 地址 model: deepseek-r1:7b # 使用的模型名称 temperature: 0.1 # 较低的温度使输出更确定适合工具调用 agent: name: LocalDeepSeekAgent max_iterations: 10 # 防止 Agent 陷入死循环 skill_timeout: 30 # 技能执行超时时间秒 skills: enabled: - file_read - file_write - get_current_time - list_directory # 可以在这里为技能配置参数如允许访问的根目录 file_root: C:\\Users\\YourName\\AgentWorkspace # Windows 路径示例3. 实现简易技能与 Agent 核心3.1 实现技能技能是实现具体功能的地方。每个技能都是一个 Python 函数并附带元数据描述。skills/file_ops.py:import os import json from datetime import datetime from pathlib import Path from typing import Dict, Any def read_file(params: Dict[str, Any]) - Dict[str, Any]: 读取指定文件的内容。 参数: {“path”: “文件路径字符串”} try: file_path Path(params.get(path)) # 简单的安全限制检查路径是否在允许的根目录下实际项目需更严格 # allowed_root Path(config.get(skills, {}).get(file_root, .)) # if not file_path.resolve().is_relative_to(allowed_root.resolve()): # return {success: False, error: Access denied: path out of allowed scope.} if not file_path.is_file(): return {success: False, error: fFile not found: {file_path}} with open(file_path, r, encodingutf-8) as f: content f.read() return {success: True, content: content} except Exception as e: return {success: False, error: str(e)} def write_file(params: Dict[str, Any]) - Dict[str, Any]: 将内容写入指定文件覆盖。 参数: {“path”: “文件路径”, “content”: “要写入的文本内容”} try: file_path Path(params.get(path)) content params.get(content, ) # 安全限制同上 file_path.parent.mkdir(parentsTrue, exist_okTrue) with open(file_path, w, encodingutf-8) as f: f.write(content) return {success: True, message: fFile written: {file_path}} except Exception as e: return {success: False, error: str(e)} def list_directory(params: Dict[str, Any]) - Dict[str, Any]: 列出指定目录下的文件和文件夹。 参数: {“path”: “目录路径” 可选默认为当前目录} try: target_path Path(params.get(path, .)) if not target_path.is_dir(): return {success: False, error: fNot a directory: {target_path}} items [] for item in target_path.iterdir(): items.append({ name: item.name, type: directory if item.is_dir() else file, size: item.stat().st_size if item.is_file() else 0 }) return {success: True, items: items} except Exception as e: return {success: False, error: str(e)} # 技能元数据用于告知模型此技能的功能和调用方式 FILE_OPS_SKILLS { read_file: { function: read_file, description: 读取一个文本文件的内容。, parameters: { path: {type: string, description: 要读取的文件的完整路径。} } }, write_file: { function: write_file, description: 将文本内容写入一个文件。如果文件不存在则创建存在则覆盖。, parameters: { path: {type: string, description: 要写入的文件的完整路径。}, content: {type: string, description: 要写入的文本内容。} } }, list_directory: { function: list_directory, description: 列出指定目录下的所有文件和子目录。, parameters: { path: {type: string, description: 要列出的目录路径。默认为当前目录。, required: False} } } }skills/system_info.py:from datetime import datetime import platform from typing import Dict, Any def get_current_time(params: Dict[str, Any]) - Dict[str, Any]: 获取当前系统时间。无需参数。 try: current_time datetime.now().strftime(%Y-%m-%d %H:%M:%S) return {success: True, current_time: current_time} except Exception as e: return {success: False, error: str(e)} def get_system_info(params: Dict[str, Any]) - Dict[str, Any]: 获取基本的系统信息。无需参数。 try: info { system: platform.system(), release: platform.release(), machine: platform.machine(), processor: platform.processor(), } return {success: True, info: info} except Exception as e: return {success: False, error: str(e)} SYSTEM_INFO_SKILLS { get_current_time: { function: get_current_time, description: 获取当前的日期和时间。, parameters: {} }, get_system_info: { function: get_system_info, description: 获取操作系统和硬件的基本信息。, parameters: {} } }3.2 实现 Agent 核心调度逻辑Agent 的核心是循环问模型 - 解析工具调用 - 执行技能 - 反馈结果 - 继续问模型。agent_core.py:import requests import json import yaml from typing import Dict, List, Any, Optional import time from skills.file_ops import FILE_OPS_SKILLS from skills.system_info import SYSTEM_INFO_SKILLS class LocalAIAgent: def __init__(self, config_path: str config.yaml): with open(config_path, r, encodingutf-8) as f: self.config yaml.safe_load(f) self.ollama_url self.config[ollama][base_url] self.model self.config[ollama][model] self.max_iterations self.config[agent][max_iterations] # 加载启用的技能 self.skills {} self._load_skills() # 对话历史 self.conversation_history [] def _load_skills(self): 根据配置加载技能。 enabled_skills self.config[skills][enabled] all_skill_dicts [FILE_OPS_SKILLS, SYSTEM_INFO_SKILLS] for skill_dict in all_skill_dicts: for skill_name, skill_info in skill_dict.items(): if skill_name in enabled_skills: self.skills[skill_name] skill_info print(fLoaded skills: {list(self.skills.keys())}) def _call_ollama(self, messages: List[Dict]) - Dict[str, Any]: 调用 Ollama API 进行聊天补全。 payload { model: self.model, messages: messages, stream: False, options: { temperature: self.config[ollama].get(temperature, 0.1) } } try: response requests.post( f{self.ollama_url}/api/chat, jsonpayload, timeout60 ) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: raise Exception(fOllama API call failed: {e}) def _extract_tool_call(self, response_message: Dict) - Optional[Dict]: 从模型回复中解析工具调用请求。 期望模型回复的 content 字段是一个 JSON 字符串格式如 {action: skill_name, params: {...}} 或者是一个包含类似结构的思考文本。 这是一个简化的解析器实际应用可能需要更复杂的逻辑或使用 OpenAI 兼容的 tool_calls 格式。 content response_message.get(content, ).strip() # 尝试从 JSON 块中解析 import re json_match re.search(r\{.*action.*\}, content, re.DOTALL) if json_match: try: tool_call json.loads(json_match.group()) if action in tool_call and params in tool_call: return tool_call except json.JSONDecodeError: pass # 如果模型直接给出了最终答案则返回 None return None def _execute_skill(self, skill_name: str, params: Dict) - Dict[str, Any]: 执行指定的技能。 if skill_name not in self.skills: return {success: False, error: fSkill {skill_name} not found or disabled.} skill_info self.skills[skill_name] skill_func skill_info[function] try: # 在实际项目中这里应加入更严格的参数验证和权限检查 result skill_func(params) return result except Exception as e: return {success: False, error: fSkill execution error: {str(e)}} def run(self, user_input: str) - str: 运行 Agent 处理用户输入。 print(f\n[User]: {user_input}) # 初始化消息历史可以加入系统提示词来引导模型使用工具 system_prompt f你是一个有帮助的AI助手可以调用工具来帮助用户。你可以使用的工具如下 {json.dumps([{name: k, description: v[description], parameters: v[parameters]} for k, v in self.skills.items()], indent2)} 当你需要调用工具时请严格按照以下JSON格式回复且只回复这个JSON对象不要添加其他解释 {{action: 工具名称, params: {{参数名: 参数值}}}} 如果用户的问题不需要调用工具或者工具调用完成后得到了最终答案请直接以自然语言回复。 messages [ {role: system, content: system_prompt}, *self.conversation_history[-6:], # 保留最近几轮历史作为上下文 {role: user, content: user_input} ] iteration 0 final_answer None while iteration self.max_iterations: iteration 1 print(f\n[Agent] Iteration {iteration}: Querying model...) # 1. 调用模型 ollama_response self._call_ollama(messages) assistant_message ollama_response[message] messages.append(assistant_message) # 将模型回复加入历史 # 2. 尝试解析工具调用 tool_call self._extract_tool_call(assistant_message) if tool_call: action tool_call[action] params tool_call[params] print(f[Agent] Detected tool call: {action} with params {params}) # 3. 执行工具 tool_result self._execute_skill(action, params) print(f[Agent] Tool result: {tool_result}) # 4. 将工具执行结果作为新的上下文反馈给模型 result_message { role: user, content: fThe result of tool {action} is: {json.dumps(tool_result)}. Please continue based on this result. } messages.append(result_message) # 继续下一轮循环让模型基于结果决定下一步 else: # 模型没有调用工具直接给出了最终答案 final_answer assistant_message[content] print(f[Agent] Model provided final answer.) break if not final_answer and iteration self.max_iterations: final_answer 已达到最大迭代次数未能完成请求。 # 将本轮完整交互加入历史可控制历史长度 self.conversation_history.extend([ {role: user, content: user_input}, {role: assistant, content: final_answer or No answer generated.} ]) return final_answer or Agent finished without a final answer.4. 运行验证与结果分析4.1 创建程序入口并运行main.py:from agent_core import LocalAIAgent def main(): agent LocalAIAgent() print(Local AI Agent 已启动。输入 quit 或 exit 退出。) while True: try: user_input input(\n ) if user_input.lower() in [quit, exit, q]: print(再见) break if not user_input.strip(): continue answer agent.run(user_input) print(f\n[Assistant]: {answer}) except KeyboardInterrupt: print(\n程序被中断。) break except Exception as e: print(f\n[Error]: 处理请求时出错 - {e}) if __name__ __main__: main()4.2 启动与测试确保 Ollama 服务运行在另一个终端执行ollama serve或确保 Ollama 后台服务已启动。运行 Agent在项目根目录下激活虚拟环境后运行python main.py看到Loaded skills: [read_file, write_file, get_current_time, list_directory]和提示符即表示成功。进行功能测试测试系统技能 现在几点了 [Agent] Iteration 1: Querying model... [Agent] Detected tool call: get_current_time with params {} [Agent] Tool result: {success: True, current_time: 2024-01-01 15:30:22} [Agent] Iteration 2: Querying model... [Agent] Model provided final answer. [Assistant]: 当前时间是 2024-01-01 15:30:22。测试文件操作技能 请在我的文档文件夹假设是 C:\Users\YourName\AgentWorkspace里创建一个名为 test_agent.txt 的文件内容写“Hello from DeepSeek Agent”。观察 Agent 的思考过程它应该会调用write_file技能。执行成功后你可以去对应目录查看文件。测试复杂任务 请列出我的文档文件夹里有什么文件然后告诉我文件数量。Agent 会先调用list_directory获取结果后模型会理解结果并计算数量最终给出回答。4.3 结果分析成功标志Agent 能正确理解你的意图将任务分解调用正确的技能并将执行结果整合成自然语言回复。模型规划能力deepseek-r1:7b这类模型在收到清晰的系统提示和工具描述后通常能较好地规划简单任务。对于多步骤任务需要依赖其推理能力。系统健壮性当前实现是一个基础原型。生产环境需要更完善的错误处理、技能执行超时控制、对话历史管理以及防止技能被滥用的安全沙箱。5. 常见问题排查与调试在整合过程中你可能会遇到以下典型问题。这里提供排查思路。问题现象可能原因检查方式处理建议启动 Agent 时报连接 Ollama 失败1. Ollama 服务未启动。2.config.yaml中的base_url端口错误。3. 防火墙或网络策略阻止连接。1. 在浏览器访问http://localhost:11434看是否返回 Ollama 信息。2. 运行ollama list命令确认服务正常。3. 检查config.yaml配置。1. 确保先运行ollama serve。2. 确认 Ollama 版本API 端口是否为默认的 11434。3. 暂时关闭防火墙或检查代理设置。模型不调用工具总是直接回答1. 系统提示词system_prompt不够清晰或格式不对。2. 模型能力有限不理解工具调用格式。3. 技能描述不够准确。1. 打印出发送给模型的完整messages检查system_prompt。2. 尝试用更简单的指令测试如“使用 get_current_time 工具”。3. 换用qwen2.5:7b或llama3.1等工具调用能力更强的模型测试。1. 优化提示词明确要求模型以指定 JSON 格式回复。2. 在提示词中提供 1-2 个清晰的调用示例。3. 考虑使用支持 OpenAI 兼容tool_calls格式的模型和前端如 LiteLLM。技能执行失败返回权限错误1. 程序运行用户没有目标文件/目录的读写权限。2. 路径不存在或拼写错误。3. Windows 路径中的反斜杠未转义。1. 检查 Agent 进程的用户权限。2. 在技能函数内打印传入的params确认路径正确。3. 使用Path库处理路径它兼容不同操作系统。1. 为 Agent 设置一个安全的工作根目录并在技能中做路径解析和限制。2. 在技能函数开头添加详细的日志和参数验证。3. 使用绝对路径并确保路径存在。Agent 陷入无限循环1. 模型反复调用同一个工具或无效工具。2.max_iterations设置过高或逻辑有误。3. 工具执行结果格式混乱模型无法理解。1. 查看每次迭代中模型输出的content和解析出的tool_call。2. 检查_extract_tool_call函数是否能正确处理非工具调用回复。1. 降低max_iterations到 5-8。2. 在_extract_tool_call中加强判断如果content看起来是自然语言答案则停止循环。3. 确保工具返回的结果是结构化的字典。处理中文时出现乱码1. 文件读写编码不是utf-8。2. 控制台或终端编码不支持 UTF-8。1. 检查技能中open函数是否指定encodingutf-8。2. 在 Python 文件开头添加# -*- coding: utf-8 -*-。3. 检查终端编码Windows CMD 需使用chcp 65001。1. 统一使用 UTF-8 编码。2. 在 Windows 上使用 PowerShell 或支持 UTF-8 的终端如 Windows Terminal。3. 对输出字符串进行必要的编码处理。6. 进阶优化与生产环境建议上述原型验证了基本流程。要将其用于更严肃的场景需要考虑以下优化方向。6.1 技能框架标准化采用成熟框架考虑使用LangChain Tools、AutoGen的UserProxyAgent或CrewAI的Tool类来定义技能。它们提供了更标准的接口、更安全的执行环境和更丰富的工具库。技能动态注册与发现实现一个技能管理器支持热加载技能模块无需重启 Agent。6.2 提示工程优化结构化输出使用 Ollama 的format参数或模型本身的微调能力强制其以 JSON 等结构化格式输出使_extract_tool_call的解析更稳定。少样本示例在system_prompt中提供 2-3 个完整的“用户问题 - 模型思考 - 工具调用 - 结果 - 最终回答”的示例大幅提升模型遵循格式的能力。6.3 安全与权限控制至关重要技能沙箱对于执行 Shell 命令 (subprocess) 或高风险操作的技能必须在沙箱环境中运行限制其可访问的资源。用户确认在执行删除文件、修改系统设置等危险操作前Agent 应主动向用户请求确认。访问控制列表为不同技能配置 ACL定义其允许访问的文件路径、网络地址和系统调用。6.4 客户端与用户体验图形界面使用PyQt、Tkinter或web技术如Gradio、Streamlit构建一个简单的 GUI方便非技术用户使用。会话持久化将conversation_history保存到数据库或文件支持多轮对话和历史回顾。流式响应对接 Ollama 的流式 API实现类似 ChatGPT 的打字机效果提升体验。6.5 解决“上网查询信息受限”问题输入材料中提到“上网查询信息经常受限”这通常是因为技能缺失没有实现一个可以发起 HTTP 请求、解析网页内容的“网络搜索”技能。网络环境本地环境可能无法直接访问外部网络或需要配置代理。解决方案实现网络技能在skills/下创建web_search.py使用requests或aiohttp库封装搜索 API如 Serper、Google Custom Search或简单的网页抓取功能。务必遵守目标网站的robots.txt和服务条款。配置代理在技能函数或全局requests会话中配置代理设置以适配公司或地区的网络环境。使用本地知识库对于内部信息查询可以集成ChromaDB、FAISS等向量数据库将本地文档嵌入后供模型检索这是更安全、可控的方案。通过以上步骤你不仅能够搭建一个可运行的本地 AI Agent 系统还能理解其内部协作机制并具备排查问题和进行生产级优化的能力。核心在于明确模型、技能、Agent 三者的边界并通过稳定的通信协议如结构化 JSON将它们连接起来。