ERNIE-4.5-0.3B-PT Chainlit前端AI工作流支持多步骤任务分解与自动执行想象一下你有一个智能助手不仅能回答简单问题还能理解复杂的多步骤任务比如“帮我分析这个季度的销售数据找出增长最快的产品然后写一份总结报告”。传统的大模型往往只能处理单轮对话而今天要介绍的这套方案让AI具备了“思考-分解-执行”的完整工作流能力。基于vLLM部署的ERNIE-4.5-0.3B-PT模型配合Chainlit构建的交互式前端我们打造了一个能够理解复杂指令、自动分解任务、并逐步执行的AI工作流系统。这不仅仅是另一个聊天机器人而是一个真正能帮你完成实际工作的智能助手。1. 为什么需要AI工作流在日常工作中我们经常遇到需要多步骤完成的任务。比如数据分析、报告撰写、代码调试等这些任务如果让人工智能来协助往往需要多次来回沟通传统方式你问一个问题AI回答你再问下一个问题AI再回答……效率低下工作流方式你描述完整任务AI自动分解步骤一步步执行最后给你完整结果这套系统的核心价值在于“自动化”和“连续性”。它让AI从被动的问答工具变成了主动的任务执行者。1.1 工作流的三个核心优势效率提升减少人工干预次数一次输入获得完整结果逻辑连贯保持任务执行的上下文一致性避免信息断层结果可靠每个步骤都有明确的目标和输出便于验证和调整2. 技术栈选型为什么是ERNIE-4.5-0.3B-PT Chainlit选择合适的技术组合是项目成功的关键。我们对比了多种方案最终确定了现在的技术栈。2.1 ERNIE-4.5-0.3B-PT小而精的文本生成专家ERNIE 4.5系列模型在技术上有很多创新特别是0.3B这个版本虽然参数规模不大但在特定任务上表现优异多模态异构MoE预训练模型同时在文本和视觉模态上训练能更好地理解多模态信息的细微差别。通过模态隔离路由等技术确保不同模态都能得到有效表示。高效推理优化支持4位/2位无损量化在保持精度的同时大幅降低资源消耗。基于PaddlePaddle框架能在各种硬件平台上提供高性能推理。特定任务优化经过监督微调和强化学习优化在语言理解和生成任务上表现稳定可靠。对于工作流场景来说0.3B的规模恰到好处——足够智能处理复杂逻辑又不会对计算资源要求过高。2.2 vLLM高性能推理引擎vLLM是目前最流行的大模型推理框架之一它的优势很明显内存效率高采用PagedAttention技术大幅减少显存占用吞吐量大支持连续批处理能同时处理多个请求部署简单提供RESTful API方便与其他系统集成用vLLM部署ERNIE-4.5-0.3B-PT能获得接近原生的推理性能。2.3 Chainlit交互式前端框架Chainlit是专门为AI应用设计的Web框架它有几个特点特别适合工作流场景对话流可视化能清晰展示多轮对话的流程组件丰富支持文件上传、代码高亮、图表展示等开发简单Python原生学习成本低最重要的是Chainlit天然支持“思考-行动”模式这正是工作流需要的。3. 快速部署10分钟搭建完整环境下面我们一步步搭建整个系统。即使你是第一次接触这些技术也能跟着完成。3.1 环境准备首先确保你的环境满足以下要求Python 3.8或更高版本至少8GB可用内存支持CUDA的GPU可选CPU也能运行但速度较慢安装必要的依赖# 创建虚拟环境 python -m venv ernie_workflow source ernie_workflow/bin/activate # Linux/Mac # 或 ernie_workflow\Scripts\activate # Windows # 安装核心依赖 pip install vllm pip install chainlit pip install paddlepaddle # ERNIE基于PaddlePaddle # 安装其他工具包 pip install requests pip install numpy3.2 部署ERNIE-4.5-0.3B-PT模型使用vLLM部署模型非常简单只需要几行命令# 启动vLLM服务 python -m vllm.entrypoints.openai.api_server \ --model ernie-4.5-0.3b-pt \ --served-model-name ernie-workflow \ --port 8000 \ --max-model-len 4096这里有几个关键参数需要说明--model指定要加载的模型这里用ERNIE-4.5-0.3B-PT--served-model-name服务名称后续调用时使用--port服务端口默认8000--max-model-len最大上下文长度4096足够大多数工作流使用启动后你会看到类似这样的输出INFO 07-15 14:30:25 llm_engine.py:72] Initializing an LLM engine with config: ... INFO 07-15 14:30:30 model_runner.py:84] Loading model weights... INFO 07-15 14:31:15 llm_engine.py:199] Model loaded successfully. INFO 07-15 14:31:15 api_server.py:107] Serving on http://0.0.0.0:80003.3 验证模型服务服务启动后我们可以用简单的方法验证是否正常# 查看服务日志 cat /root/workspace/llm.log如果看到“Model loaded successfully”和“Serving on”这样的信息说明模型已经成功加载并开始服务。也可以用curl测试一下curl http://localhost:8000/v1/models应该返回类似这样的JSON{ object: list, data: [ { id: ernie-workflow, object: model, created: 1678888888, owned_by: vllm } ] }4. 构建Chainlit前端应用现在模型服务已经就绪我们来构建用户交互界面。4.1 创建Chainlit应用新建一个文件workflow_app.pyimport chainlit as cl import requests import json from typing import List, Dict, Any # vLLM服务的地址 VLLM_API_URL http://localhost:8000/v1/chat/completions class TaskWorkflow: 任务工作流处理器 def __init__(self): self.conversation_history [] def call_ernie_model(self, messages: List[Dict]) - str: 调用ERNIE模型API headers { Content-Type: application/json } payload { model: ernie-workflow, messages: messages, temperature: 0.7, max_tokens: 1024 } try: response requests.post(VLLM_API_URL, headersheaders, datajson.dumps(payload), timeout30) response.raise_for_status() result response.json() return result[choices][0][message][content] except Exception as e: return f调用模型失败: {str(e)} def analyze_task(self, user_input: str) - List[str]: 分析用户输入分解为子任务 system_prompt 你是一个任务分解专家。请将用户的任务分解为具体的执行步骤。 每个步骤应该清晰、可执行。输出格式1. 步骤描述\n2. 步骤描述\n... messages [ {role: system, content: system_prompt}, {role: user, content: user_input} ] decomposition self.call_ernie_model(messages) # 解析分解结果 steps [step.strip() for step in decomposition.split(\n) if step.strip() and step[0].isdigit()] return steps def execute_step(self, step: str, context: Dict[str, Any]) - str: 执行单个步骤 system_prompt f你正在执行任务的一个步骤。当前上下文{json.dumps(context, ensure_asciiFalse)} 请专注于完成这个步骤{step} messages [ {role: system, content: system_prompt}, {role: user, content: f请执行{step}} ] return self.call_ernie_model(messages) # 创建全局工作流实例 workflow TaskWorkflow() cl.on_chat_start async def start_chat(): 聊天开始时的初始化 await cl.Message( content你好我是你的AI工作流助手。我可以帮你分解复杂任务并逐步执行。请告诉我你想做什么 ).send() cl.on_message async def handle_message(message: cl.Message): 处理用户消息 user_input message.content # 显示思考过程 thinking_msg cl.Message(content正在分析你的任务...) await thinking_msg.send() # 步骤1任务分解 steps workflow.analyze_task(user_input) if not steps: await cl.Message(content无法分解任务请尝试更清晰的描述。).send() return # 显示分解结果 steps_text 我将按以下步骤执行\n\n \n.join([f{i1}. {step} for i, step in enumerate(steps)]) await cl.Message(contentsteps_text).send() # 步骤2逐步执行 context {original_task: user_input, steps: steps} results [] for i, step in enumerate(steps, 1): step_msg cl.Message(contentf正在执行步骤 {i}/{len(steps)}: {step}) await step_msg.send() result workflow.execute_step(step, context) results.append(result) # 更新上下文 context[fstep_{i}_result] result # 显示步骤结果 await cl.Message(contentf步骤 {i} 完成\n{result}).send() # 步骤3汇总结果 summary_msg cl.Message(content正在生成最终总结...) await summary_msg.send() final_summary workflow.generate_summary(user_input, steps, results) await cl.Message(contentfinal_summary).send() cl.password_auth_callback def auth_callback(username: str, password: str): 简单的认证回调可选 # 在实际应用中这里应该连接你的用户数据库 if username admin and password admin123: return cl.User(identifieradmin) else: return None4.2 配置Chainlit创建chainlit.md文件作为应用说明# AI工作流助手 这是一个基于ERNIE-4.5-0.3B-PT的智能工作流系统能够 1. **理解复杂任务**将多步骤任务自动分解 2. **逐步执行**按逻辑顺序完成每个子任务 3. **保持上下文**在整个流程中维护任务状态 4. **生成总结**提供完整的执行报告 ## 使用示例 你可以尝试以下类型的任务 - 帮我分析这个季度的销售数据找出增长最快的产品然后写一份总结报告 - 写一个Python爬虫抓取新闻网站的头条新闻然后分析情感倾向 - 设计一个用户注册系统包括前端界面、后端API和数据库设计 ## 注意事项 - 每个步骤执行时间可能较长请耐心等待 - 复杂任务可能需要多次交互 - 结果仅供参考重要决策请人工复核创建chainlit.yaml配置文件chainlit: 1.0.0 # 应用配置 app: title: AI工作流助手 description: 基于ERNIE-4.5的多步骤任务执行系统 icon: https://cdn-icons-png.flaticon.com/512/2103/2103655.png theme: primaryColor: #4f46e5 backgroundColor: #ffffff # 功能配置 features: - name: file_upload accepted_files: - .txt - .pdf - .csv - .json - name: human_feedback - name: speech_to_text # 模型配置 model: provider: vllm name: ernie-4.5-0.3b-pt parameters: temperature: 0.7 max_tokens: 1024 # UI配置 ui: show_thinking: true show_step_by_step: true auto_scroll: true4.3 启动Chainlit应用现在可以启动前端应用了# 启动Chainlit chainlit run workflow_app.py -w --port 7860参数说明-w启用自动重载开发时很有用--port指定端口默认7860启动成功后打开浏览器访问http://localhost:7860就能看到Chainlit的交互界面了。5. 实际应用案例展示让我们通过几个具体例子看看这个系统在实际工作中能做什么。5.1 案例一数据分析与报告生成用户输入 “帮我分析一下最近三个月的销售数据找出销售额最高的产品类别然后写一份分析报告包括趋势分析和改进建议。”系统分解的步骤理解销售数据的结构和时间范围计算各产品类别的销售额识别销售额最高的类别分析销售趋势月度变化提出具体的改进建议撰写完整的分析报告执行过程 系统会依次执行每个步骤。比如在步骤2它会模拟计算过程在步骤4它会分析增长趋势最后生成一份包含数据、分析和建议的完整报告。最终输出 一份结构清晰的报告包括数据概览表格趋势分析图表描述关键发现总结具体可操作的建议5.2 案例二代码开发任务用户输入 “开发一个简单的待办事项应用需要前端界面、后端API和数据库设计。”系统分解的步骤设计数据库表结构用户、任务、分类等创建后端API任务增删改查、用户认证设计前端界面任务列表、添加表单、筛选功能实现前后端连接编写使用说明执行过程 系统会为每个步骤生成具体的代码。比如步骤1生成SQL建表语句步骤2生成Python Flask代码步骤3生成HTML/CSS/JavaScript代码。最终输出 一个完整的可运行应用代码包包含数据库schema文件后端API代码前端界面代码部署和运行说明5.3 案例三内容创作工作流用户输入 “为我们的新产品写一篇推广文章需要包括产品介绍、功能亮点、用户评价和购买引导。”系统分解的步骤收集产品基本信息提炼核心功能亮点模拟用户评价和反馈设计文章结构和大纲撰写各个部分内容优化语言和排版添加购买引导和行动号召执行过程 系统会像专业文案一样工作先做调研再定结构然后填充内容最后优化润色。最终输出 一篇完整的推广文章包括吸引人的标题和开头清晰的产品介绍有说服力的功能描述真实的用户评价模拟明确的购买引导专业的排版建议6. 高级功能与定制基础的工作流已经很有用但我们还可以进一步扩展功能。6.1 支持文件上传和处理Chainlit天然支持文件上传我们可以扩展工作流来处理上传的文件cl.on_message async def handle_message_with_files(message: cl.Message): 处理带文件的消息 # 检查是否有上传的文件 if message.elements: files_info [] for element in message.elements: if hasattr(element, path): # 处理文件内容 file_content await process_uploaded_file(element.path) files_info.append({ name: element.name, type: element.type, content: file_content[:500] # 只取前500字符 }) # 将文件信息添加到用户输入中 enhanced_input f{message.content}\n\n上传的文件信息{json.dumps(files_info, ensure_asciiFalse)} # 继续原来的处理流程 await handle_workflow(enhanced_input) else: await handle_workflow(message.content) async def process_uploaded_file(file_path: str) - str: 处理上传的文件 # 根据文件类型选择处理方式 if file_path.endswith(.txt): with open(file_path, r, encodingutf-8) as f: return f.read() elif file_path.endswith(.csv): # 处理CSV文件 import pandas as pd df pd.read_csv(file_path) return df.head().to_string() elif file_path.endswith(.json): with open(file_path, r, encodingutf-8) as f: data json.load(f) return json.dumps(data, ensure_asciiFalse, indent2) else: return 文件类型暂不支持详细解析6.2 添加记忆功能让系统记住之前的对话历史实现更连贯的交互class MemoryEnhancedWorkflow(TaskWorkflow): 带记忆功能的工作流 def __init__(self): super().__init__() self.conversation_memory [] self.max_memory_length 10 # 记住最近10轮对话 def add_to_memory(self, role: str, content: str): 添加对话到记忆 self.conversation_memory.append({ role: role, content: content, timestamp: time.time() }) # 保持记忆长度 if len(self.conversation_memory) self.max_memory_length: self.conversation_memory self.conversation_memory[-self.max_memory_length:] def get_context_from_memory(self) - str: 从记忆中提取上下文 if not self.conversation_memory: return context_parts [] for i, msg in enumerate(self.conversation_memory[-5:], 1): # 最近5条 context_parts.append(f{msg[role]}: {msg[content][:100]}...) return 最近的对话历史\n \n.join(context_parts)6.3 实现条件分支逻辑复杂的工作流可能需要根据中间结果选择不同的执行路径def execute_conditional_workflow(self, user_input: str): 执行带条件分支的工作流 # 第一步分析任务并制定计划 plan self.create_execution_plan(user_input) results {} context {original_task: user_input} # 按计划执行 for step in plan[steps]: step_result self.execute_step(step[description], context) results[step[id]] step_result context[step[id]] step_result # 检查是否需要分支 if step.get(condition): condition_met self.evaluate_condition(step[condition], context) if condition_met: # 执行分支步骤 branch_steps step[branch_steps] for branch_step in branch_steps: branch_result self.execute_step(branch_step, context) results[f{step[id]}_branch] branch_result return self.generate_final_output(plan, results) def evaluate_condition(self, condition: str, context: Dict) - bool: 评估条件是否满足 # 简单的条件评估逻辑 # 在实际应用中这里可以更复杂 condition_prompt f根据以下上下文判断条件是否成立 上下文{json.dumps(context, ensure_asciiFalse)} 条件{condition} 只回答是或否。 response self.call_ernie_model([ {role: user, content: condition_prompt} ]) return 是 in response.lower()7. 性能优化与最佳实践要让工作流系统运行得更稳定高效有几个关键点需要注意。7.1 模型调用优化批量处理如果有多个相似任务可以批量发送给模型def batch_process_tasks(self, tasks: List[str]) - List[str]: 批量处理任务 batch_messages [] for task in tasks: batch_messages.append([ {role: system, content: 请处理以下任务}, {role: user, content: task} ]) # 这里需要vLLM支持批量调用 # 实际实现可能因API而异 results [] for messages in batch_messages: result self.call_ernie_model(messages) results.append(result) return results缓存机制对常见任务的结果进行缓存from functools import lru_cache import hashlib class CachedWorkflow(TaskWorkflow): lru_cache(maxsize100) def cached_call(self, prompt_hash: str, messages: List[Dict]) - str: 带缓存的模型调用 # 实际调用模型 return self.call_ernie_model(messages) def get_prompt_hash(self, messages: List[Dict]) - str: 生成提示词的哈希值 prompt_str json.dumps(messages, sort_keysTrue) return hashlib.md5(prompt_str.encode()).hexdigest() def smart_call(self, messages: List[Dict]) - str: 智能调用先查缓存 prompt_hash self.get_prompt_hash(messages) # 检查缓存 if hasattr(self, _cache) and prompt_hash in self._cache: return self._cache[prompt_hash] # 调用模型 result self.call_ernie_model(messages) # 更新缓存 if not hasattr(self, _cache): self._cache {} self._cache[prompt_hash] result return result7.2 错误处理与重试网络调用和模型推理都可能出错需要有健壮的错误处理def robust_model_call(self, messages: List[Dict], max_retries: int 3) - str: 健壮的模型调用支持重试 for attempt in range(max_retries): try: response requests.post( VLLM_API_URL, headers{Content-Type: application/json}, datajson.dumps({ model: ernie-workflow, messages: messages, temperature: 0.7, max_tokens: 1024 }), timeout30 ) response.raise_for_status() result response.json() return result[choices][0][message][content] except requests.exceptions.Timeout: if attempt max_retries - 1: await asyncio.sleep(2 ** attempt) # 指数退避 continue else: return 请求超时请稍后重试 except requests.exceptions.RequestException as e: if attempt max_retries - 1: await asyncio.sleep(1) continue else: return f网络错误{str(e)} except Exception as e: return f处理错误{str(e)} return 请求失败请检查网络连接7.3 资源监控与管理对于生产环境需要监控系统资源使用情况import psutil import time class ResourceMonitor: 资源监控器 def __init__(self): self.start_time time.time() self.request_count 0 def log_request(self): 记录请求 self.request_count 1 def get_status(self) - Dict: 获取系统状态 cpu_percent psutil.cpu_percent(interval1) memory_info psutil.virtual_memory() disk_usage psutil.disk_usage(/) uptime time.time() - self.start_time return { uptime_seconds: uptime, uptime_human: self.format_time(uptime), total_requests: self.request_count, requests_per_minute: self.request_count / (uptime / 60), cpu_percent: cpu_percent, memory_percent: memory_info.percent, memory_used_gb: memory_info.used / (1024**3), disk_percent: disk_usage.percent, timestamp: time.strftime(%Y-%m-%d %H:%M:%S) } def format_time(self, seconds: float) - str: 格式化时间 hours int(seconds // 3600) minutes int((seconds % 3600) // 60) secs int(seconds % 60) return f{hours}h {minutes}m {secs}s8. 部署与运维建议8.1 生产环境部署对于生产环境建议采用以下架构用户请求 → Nginx反向代理 → Chainlit应用 → vLLM服务 → ERNIE模型 ↓ 监控系统 ↓ 日志系统关键配置使用Gunicorn运行Chainlitgunicorn -w 4 -k uvicorn.workers.UvicornWorker workflow_app:app配置Nginx反向代理server { listen 80; server_name your-domain.com; location / { proxy_pass http://localhost:7860; proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection upgrade; proxy_set_header Host $host; } }设置系统服务使用systemd# /etc/systemd/system/ernie-workflow.service [Unit] DescriptionERNIE Workflow Service Afternetwork.target [Service] Useryour_user WorkingDirectory/path/to/your/app ExecStart/path/to/venv/bin/chainlit run workflow_app.py --port 7860 Restartalways [Install] WantedBymulti-user.target8.2 监控与日志完善的监控能帮助快速发现问题import logging from logging.handlers import RotatingFileHandler def setup_logging(): 配置日志系统 logger logging.getLogger(ernie_workflow) logger.setLevel(logging.INFO) # 文件处理器 file_handler RotatingFileHandler( workflow.log, maxBytes10*1024*1024, # 10MB backupCount5 ) file_handler.setLevel(logging.INFO) # 控制台处理器 console_handler logging.StreamHandler() console_handler.setLevel(logging.WARNING) # 格式 formatter logging.Formatter( %(asctime)s - %(name)s - %(levelname)s - %(message)s ) file_handler.setFormatter(formatter) console_handler.setFormatter(formatter) logger.addHandler(file_handler) logger.addHandler(console_handler) return logger # 使用示例 logger setup_logging() cl.on_message async def handle_message_with_logging(message: cl.Message): 带日志记录的消息处理 logger.info(f收到消息: {message.content[:100]}...) try: # 处理消息 result await process_message(message) logger.info(消息处理成功) return result except Exception as e: logger.error(f消息处理失败: {str(e)}) raise8.3 安全考虑API密钥管理不要硬编码密钥使用环境变量输入验证验证用户输入防止注入攻击速率限制防止滥用数据隔离不同用户的数据要隔离import os from dotenv import load_dotenv # 加载环境变量 load_dotenv() class SecureWorkflow(TaskWorkflow): def __init__(self): super().__init__() self.api_key os.getenv(VLLM_API_KEY) self.rate_limiter RateLimiter(max_calls100, period60) # 每分钟100次 def validate_input(self, user_input: str) - bool: 验证用户输入 # 检查长度 if len(user_input) 5000: return False # 检查敏感内容 sensitive_keywords [恶意关键词1, 恶意关键词2] for keyword in sensitive_keywords: if keyword in user_input.lower(): return False return True def call_ernie_model(self, messages: List[Dict]) - str: 安全的模型调用 # 检查速率限制 if not self.rate_limiter.allow(): return 请求过于频繁请稍后再试 # 添加API密钥 headers { Content-Type: application/json, Authorization: fBearer {self.api_key} } # 其余代码...9. 总结通过ERNIE-4.5-0.3B-PT模型和Chainlit框架我们构建了一个强大的AI工作流系统。这个系统不仅能够理解复杂的多步骤任务还能自动分解并执行这些任务大大提高了工作效率。9.1 核心价值回顾自动化程度高从任务理解到执行完成全程自动化逻辑连贯性强保持任务执行的上下文一致性扩展性好支持文件处理、条件分支、记忆功能等高级特性部署简单基于成熟的开源框架部署和维护都很方便9.2 适用场景这个系统特别适合以下场景数据分析与报告自动处理数据并生成分析报告内容创作从大纲到成文的完整写作流程代码开发需求分析、设计、实现、测试的完整流程业务流程标准化的多步骤业务处理学习辅导分步骤的教学和练习指导9.3 后续优化方向虽然现在的系统已经功能完整但还有进一步优化的空间模型微调针对特定领域的工作流进行模型微调可视化增强更丰富的进度展示和结果可视化集成扩展与其他工具和系统的深度集成性能优化更高效的模型调用和资源管理最重要的是这个系统展示了AI从简单的问答工具向智能工作伙伴演进的可能性。随着模型能力的不断提升未来AI将能在更多复杂任务中扮演关键角色。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。