不止于聊天：用Ollama API和Python打造你的第一个AI小工具

不止于聊天用Ollama API和Python打造你的第一个AI小工具当大多数人还在用大语言模型进行简单对话时聪明的开发者已经将这些能力转化为生产力工具。想象一下每天重复的代码注释工作可以自动完成海量技术文档能即时问答会议录音能自动生成摘要——这些都不需要依赖云端服务用本地运行的Ollama模型和几十行Python代码就能实现。本文将带你突破聊天机器人的思维定式通过三个实战项目演示如何将Ollama的API能力转化为解决实际问题的工具。我们不仅会覆盖API调用的核心技术要点还会教你如何设计健壮的模块化代码让你的AI工具像乐高积木一样可以自由组合扩展。1. 环境准备与API基础1.1 快速搭建Ollama环境虽然Ollama以安装简单著称但在实际项目中有几个关键配置会影响后续开发体验。推荐使用Docker运行Ollama服务既能隔离环境又便于迁移# 使用官方Docker镜像 docker run -d -p 11434:11434 --name ollama -v ollama_data:/root/.ollama ollama/ollama # 下载常用工作模型中文场景推荐 docker exec ollama ollama pull qwen:7b验证服务是否正常运行import requests response requests.get(http://localhost:11434/api/tags) print(response.json())应该能看到类似输出{ models: [ { name: qwen:7b, modified_at: 2024-05-20T08:00:00Z, size: 3820000000 } ] }1.2 API调用核心模式Ollama提供两种API调用方式适用于不同场景同步调用- 适合快速测试和短文本生成def simple_generate(prompt, modelqwen:7b): response requests.post( http://localhost:11434/api/generate, json{model: model, prompt: prompt} ) return response.json()[response]流式调用- 处理长文本时的内存优化方案def stream_generate(prompt, modelqwen:7b): response requests.post( http://localhost:11434/api/generate, json{model: model, prompt: prompt, stream: True}, streamTrue ) for chunk in response.iter_lines(): if chunk: data json.loads(chunk.decode()) yield data.get(response, )提示流式调用时建议设置超时参数避免长时间无响应阻塞程序2. 实战项目一智能代码注释生成器2.1 项目架构设计我们将构建一个能自动分析Python函数并生成专业注释的工具其工作流程如下解析Python文件提取函数定义构造LLM提示词模板处理API响应并回写文件核心模块划分code_annotator/ ├── __init__.py ├── parser.py # 代码解析 ├── prompts.py # 提示词工程 └── writer.py # 文件操作2.2 代码解析实现使用ast模块精准提取函数信息import ast def extract_functions(filepath): with open(filepath) as f: tree ast.parse(f.read()) functions [] for node in ast.walk(tree): if isinstance(node, ast.FunctionDef): func_info { name: node.name, args: [arg.arg for arg in node.args.args], lineno: node.lineno, docstring: ast.get_docstring(node) } functions.append(func_info) return functions2.3 提示词工程设计有效的提示词是获得高质量输出的关键def build_function_prompt(func_info): return f你是一名资深Python开发者请为以下函数编写专业注释 函数定义 def {func_info[name]}({, .join(func_info[args])}): ... 要求 1. 用Google风格文档字符串格式 2. 包含参数说明、返回值和示例 3. 使用中文描述 4. 保持技术准确性2.4 完整工作流集成将各模块组合成完整工具def annotate_file(filepath, modelqwen:7b): functions extract_functions(filepath) with open(filepath, r) as f: lines f.readlines() for func in reversed(functions): # 从后往前修改避免行号错位 prompt build_function_prompt(func) comment simple_generate(prompt, model) insert_index func[lineno] - 1 lines.insert(insert_index, f\n{comment}\n) f.seek(0) f.writelines(lines)使用示例python -m code_annotator annotate my_script.py3. 实战项目二本地知识库问答系统3.1 文档预处理流水线构建高效问答系统的关键在于文档预处理from pathlib import Path import PyPDF2 def chunk_document(filepath, chunk_size500): if filepath.suffix .pdf: text extract_pdf_text(filepath) else: text filepath.read_text() words text.split() chunks [ .join(words[i:ichunk_size]) for i in range(0, len(words), chunk_size) ] return chunks def extract_pdf_text(filepath): text with open(filepath, rb) as f: reader PyPDF2.PdfReader(f) for page in reader.pages: text page.extract_text() return text3.2 向量检索与问答集成使用FAISS实现本地向量检索import numpy as np from sentence_transformers import SentenceTransformer class VectorStore: def __init__(self): self.model SentenceTransformer(paraphrase-multilingual-MiniLM-L12-v2) self.index None self.chunks [] def build_index(self, documents): self.chunks [] for doc in documents: self.chunks.extend(chunk_document(doc)) embeddings self.model.encode(self.chunks) self.index FAISS.IndexFlatL2(embeddings.shape[1]) self.index.add(embeddings) def search(self, query, k3): query_embed self.model.encode([query]) distances, indices self.index.search(query_embed, k) return [self.chunks[i] for i in indices[0]]问答接口实现def answer_question(question, vector_store, modelqwen:7b): context \n.join(vector_store.search(question)) prompt f基于以下上下文回答问题 {context} 问题{question} 要求 1. 答案必须来自上下文 2. 如果上下文不包含答案回复未找到相关信息 3. 使用中文回答 return stream_generate(prompt, model)4. 高级技巧与性能优化4.1 异步批量处理当需要处理大量请求时同步调用会成为性能瓶颈import aiohttp async def batch_generate(prompts, modelqwen:7b): async with aiohttp.ClientSession() as session: tasks [] for prompt in prompts: task session.post( http://localhost:11434/api/generate, json{model: model, prompt: prompt} ) tasks.append(task) responses await asyncio.gather(*tasks) return [await r.json() for r in responses]4.2 模型参数调优不同任务需要不同的生成参数组合任务类型temperaturetop_pmax_tokens适用场景代码生成0.20.91024需要确定性输出创意写作0.70.952048鼓励多样性技术问答0.30.85512平衡准确性与流畅度文本摘要0.50.9256关键信息提取实践示例def optimized_generate(prompt, task_typecode): params { code: {temperature: 0.2, top_p: 0.9}, creative: {temperature: 0.7, top_p: 0.95} }.get(task_type, {}) response requests.post( http://localhost:11434/api/generate, json{model: qwen:7b, prompt: prompt, **params} ) return response.json()4.3 错误处理与重试机制健壮的生产级应用需要完善的错误处理from tenacity import retry, stop_after_attempt, wait_exponential retry( stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10) ) def reliable_generate(prompt): try: response requests.post( http://localhost:11434/api/generate, json{model: qwen:7b, prompt: prompt}, timeout30 ) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: print(fAPI调用失败: {str(e)}) raise5. 项目三自动化会议纪要生成器5.1 语音转文本预处理使用开源工具实现语音识别# 安装必要依赖 pip install SpeechRecognition pydub音频处理核心代码import speech_recognition as sr def transcribe_audio(audio_path): recognizer sr.Recognizer() with sr.AudioFile(audio_path) as source: audio recognizer.record(source) try: text recognizer.recognize_whisper( audio, languagezh-CN, modelsmall ) return text except sr.UnknownValueError: print(无法识别音频) return 5.2 摘要生成策略分层摘要生成提高质量首先生成章节级摘要然后基于章节摘要生成全局摘要最后提取关键行动项实现代码def generate_summary(text, modelqwen:7b): # 分段处理长文本 chunks [text[i:i2000] for i in range(0, len(text), 2000)] section_summaries [] for chunk in chunks: prompt f请用中文总结以下会议内容的关键点 {chunk} 要求 1. 不超过3个要点 2. 保留重要数据和决策 3. 使用简洁的商务语言 summary simple_generate(prompt, model) section_summaries.append(summary) # 生成最终摘要 combined \n\n.join(section_summaries) final_prompt f基于以下分段摘要生成完整的会议纪要 {combined} 要求 1. 按背景、讨论、决策结构组织 2. 标注重要时间点和责任人 3. 限制在300字以内 return simple_generate(final_prompt, model)5.3 行动项提取使用特定提示词提取可执行任务def extract_action_items(text, modelqwen:7b): prompt f从以下会议记录中提取具体行动项 {text} 要求 1. 每条行动项包含责任人、截止时间、具体任务三个要素 2. 使用Markdown表格格式输出 3. 确保每个任务都可执行和验证 response simple_generate(prompt, model) return response示例输出| 责任人 | 截止时间 | 具体任务 | |--------|------------|------------------------------| | 张三 | 2024-06-01 | 完成新API接口设计文档 | | 李四 | 2024-05-25 | 协调测试团队安排压力测试 |