当前位置：首页 > news >正文

Phi-3-mini-4k-instruct-gguf部署案例：Airflow中集成Phi-3-mini实现定时摘要任务

news 2026/6/2 13:54:31

Phi-3-mini-4k-instruct-gguf部署案例：Airflow中集成Phi-3-mini实现定时摘要任务

1. 项目背景与需求

在日常工作中，我们经常需要处理大量文本信息并生成摘要。传统的人工摘要方式效率低下，特别是在需要定时处理大量文档的场景下。本文将介绍如何在Airflow工作流中集成Phi-3-mini-4k-instruct-gguf模型，实现自动化文本摘要任务。

Phi-3-mini-4k-instruct-gguf是微软推出的轻量级文本生成模型，特别适合问答、文本改写和摘要整理等场景。它的GGUF格式版本可以高效地在本地运行，不需要依赖云端服务。

2. 环境准备与部署

2.1 基础环境要求

在开始之前，请确保你的系统满足以下要求：

Python 3.8或更高版本
已安装Airflow 2.0+
支持CUDA的GPU（推荐）或性能足够的CPU
至少8GB可用内存

2.2 模型部署

首先下载并部署Phi-3-mini-4k-instruct-gguf模型：

# 创建模型目录 mkdir -p ~/ai-models/microsoft cd ~/ai-models/microsoft # 下载模型文件 wget https://huggingface.co/TheBloke/Phi-3-mini-4k-instruct-gguf/resolve/main/phi-3-mini-4k-instruct.Q4_K_M.gguf # 安装llama-cpp-python pip install llama-cpp-python[server]

3. 创建Airflow DAG

3.1 基础DAG结构

我们将创建一个名为text_summarization的DAG，每天凌晨2点自动运行：

from datetime import datetime, timedelta from airflow import DAG from airflow.operators.python import PythonOperator default_args = { 'owner': 'airflow', 'depends_on_past': False, 'start_date': datetime(2023, 1, 1), 'retries': 1, 'retry_delay': timedelta(minutes=5), } dag = DAG( 'text_summarization', default_args=default_args, description='自动文本摘要任务', schedule_interval='0 2 * * *', catchup=False )

3.2 模型调用函数

创建一个Python函数来调用Phi-3-mini模型生成摘要：

from llama_cpp import Llama def generate_summary(text): # 初始化模型 llm = Llama( model_path="~/ai-models/microsoft/phi-3-mini-4k-instruct.Q4_K_M.gguf", n_ctx=2048, n_threads=4 ) # 构建提示词 prompt = f"请为以下文本生成一个简洁的摘要，不超过100字：\n\n{text}" # 生成摘要 output = llm( prompt, max_tokens=128, temperature=0.2, top_p=0.9, echo=False ) return output['choices'][0]['text'].strip()

4. 完整任务实现

4.1 从数据库获取文本

假设我们的文本存储在PostgreSQL数据库中：

from airflow.providers.postgres.hooks.postgres import PostgresHook def fetch_texts(**kwargs): postgres_hook = PostgresHook(postgres_conn_id='postgres_default') conn = postgres_hook.get_conn() cursor = conn.cursor() cursor.execute("SELECT id, content FROM documents WHERE summary IS NULL LIMIT 10") texts = cursor.fetchall() kwargs['ti'].xcom_push(key='texts_to_summarize', value=texts) cursor.close() conn.close()

4.2 生成并存储摘要

def process_summaries(**kwargs): ti = kwargs['ti'] texts = ti.xcom_pull(task_ids='fetch_texts', key='texts_to_summarize') postgres_hook = PostgresHook(postgres_conn_id='postgres_default') conn = postgres_hook.get_conn() cursor = conn.cursor() for doc_id, content in texts: summary = generate_summary(content) cursor.execute( "UPDATE documents SET summary = %s WHERE id = %s", (summary, doc_id) ) conn.commit() cursor.close() conn.close()

4.3 完整DAG定义

将所有任务组合起来：

fetch_task = PythonOperator( task_id='fetch_texts', python_callable=fetch_texts, provide_context=True, dag=dag, ) summarize_task = PythonOperator( task_id='process_summaries', python_callable=process_summaries, provide_context=True, dag=dag, ) fetch_task >> summarize_task

5. 性能优化与监控

5.1 模型加载优化

为了避免每次任务都重新加载模型，我们可以使用Airflow的@task.python装饰器并设置pool参数：

from airflow.decorators import task @task.python(pool="phi3_pool", pool_slots=1) def generate_summary_task(text): # 使用全局变量保持模型加载状态 if 'llm' not in globals(): globals()['llm'] = Llama( model_path="~/ai-models/microsoft/phi-3-mini-4k-instruct.Q4_K_M.gguf", n_ctx=2048, n_threads=4 ) prompt = f"摘要任务：{text}" output = globals()['llm'](prompt, max_tokens=128, temperature=0.2) return output['choices'][0]['text'].strip()

5.2 监控与日志

添加监控指标和日志记录：

from airflow.utils.log.logging_mixin import LoggingMixin def process_summaries_with_logging(**kwargs): logger = LoggingMixin().log ti = kwargs['ti'] texts = ti.xcom_pull(task_ids='fetch_texts', key='texts_to_summarize') logger.info(f"开始处理 {len(texts)} 个文档的摘要任务") for doc_id, content in texts: try: start_time = time.time() summary = generate_summary(content) duration = time.time() - start_time logger.info(f"文档 {doc_id} 摘要生成完成，耗时 {duration:.2f} 秒") # 存储摘要... except Exception as e: logger.error(f"处理文档 {doc_id} 时出错: {str(e)}") continue

6. 实际应用建议

6.1 参数调优经验

根据实际使用经验，以下参数组合效果较好：

摘要任务：temperature=0.2, max_tokens=128
文本改写：temperature=0.3, max_tokens=256
问答任务：temperature=0.1, max_tokens=512

6.2 错误处理策略

建议在DAG中添加错误处理任务：

from airflow.operators.email import EmailOperator error_email = EmailOperator( task_id='send_error_email', to='your_email@example.com', subject='摘要任务失败通知', html_content="""<p>摘要任务执行失败，请检查日志。</p>""", trigger_rule='one_failed', dag=dag ) summarize_task >> error_email