ERNIE-4.5-0.3B-PT开源镜像实战：构建支持插件调用的AI Agent平台

你是不是也遇到过这样的问题：想用AI模型做点有趣的事情，但发现部署复杂、调用麻烦，更别提让AI模型去调用外部工具和插件了。今天我要分享的，就是如何用ERNIE-4.5-0.3B-PT这个开源镜像，快速搭建一个支持插件调用的AI Agent平台。

想象一下，你只需要几分钟时间，就能拥有一个可以对话、能理解你需求、还能调用各种工具帮你完成任务的AI助手。无论是查询天气、搜索信息，还是处理文档、生成内容，都能通过简单的对话来完成。这就是我们今天要实现的。

1. 为什么选择ERNIE-4.5-0.3B-PT？

在开始动手之前，我们先简单了解一下这个模型的特点。ERNIE-4.5-0.3B-PT是一个基于MoE（混合专家）架构的轻量级模型，虽然参数只有3亿，但性能表现相当不错。

1.1 模型的核心优势

这个模型有几个让我特别喜欢的地方：

轻量高效：0.3B的参数规模意味着它可以在普通的硬件上流畅运行，不需要昂贵的GPU就能获得不错的响应速度。

多模态理解：虽然我们主要用它做文本生成，但它的底层架构支持多模态理解，这意味着未来扩展视觉、语音等功能会更容易。

插件友好：模型经过专门优化，对工具调用和插件集成的支持很好，这正是我们构建AI Agent平台所需要的。

开源免费：完全开源，没有任何使用限制，这对于个人开发者和小团队来说太重要了。

1.2 技术架构亮点

ERNIE 4.5系列模型在技术上做了很多创新：

• 异构MoE架构：不同专家处理不同任务，效率更高
• 高效推理优化：支持4位/2位量化，推理速度更快
• 多模态联合训练：文本和视觉信息能更好地协同工作

不过作为使用者，我们不需要深入了解这些技术细节，只需要知道它好用、够用就行。

2. 快速部署：从零到一的完整过程

好了，理论说太多没用，我们直接上手。整个部署过程比你想的要简单得多。

2.1 环境准备与一键部署

首先，你需要有一个可以运行Docker的环境。如果你用的是云服务商的镜像市场，通常已经预置好了所有环境。这里我以常见的Linux环境为例。

检查环境：

# 检查Docker是否安装
docker --version

# 检查Python版本（需要3.8以上）
python3 --version

如果环境没问题，我们就可以开始部署了。ERNIE-4.5-0.3B-PT镜像已经预装了所有依赖，包括vLLM推理引擎和Chainlit前端。

2.2 验证部署是否成功

部署完成后，我们需要确认服务是否正常运行。打开终端，执行以下命令：

# 查看模型服务日志
cat /root/workspace/llm.log

如果看到类似下面的输出，说明部署成功了：

INFO 07-10 14:30:25 llm_engine.py:73] Initializing an LLM engine...
INFO 07-10 14:30:25 model_runner.py:84] Loading model weights...
INFO 07-10 14:30:28 model_runner.py:121] Model loaded successfully.
INFO 07-10 14:30:28 llm_engine.py:158] LLM engine initialized.

这个过程通常需要1-2分钟，具体时间取决于你的硬件配置。模型加载完成后，就可以开始使用了。

3. 使用Chainlit构建交互式前端

现在模型服务已经跑起来了，但我们还需要一个好看、好用的界面。这就是Chainlit的用武之地。

3.1 打开Chainlit前端

在浏览器中访问Chainlit的地址（通常是http://你的服务器IP:8000），你会看到一个简洁的聊天界面。

界面分为三个主要区域：

• 左侧：对话历史列表
• 中间：主聊天区域
• 右侧：设置和插件面板

第一次打开时，界面是空的。别担心，我们马上让它变得有用起来。

3.2 进行第一次对话

在底部的输入框中，尝试问一些简单的问题：

你好，介绍一下你自己

模型会回复类似这样的内容：

你好！我是ERNIE-4.5-0.3B-PT，一个基于百度文心大模型开发的AI助手。我擅长文本生成、问答对话、代码编写等多种任务。虽然我的参数规模不大，但经过优化训练，在理解和生成中文内容方面表现不错。有什么我可以帮你的吗？

看到回复了吗？这意味着你的AI助手已经可以正常工作了。不过，这还只是个开始。真正的价值在于让这个AI能够调用外部工具。

4. 构建插件系统：让AI学会使用工具

这是最有趣的部分。我们要教AI模型如何使用各种插件，就像给一个人配备各种工具一样。

4.1 理解插件调用的基本原理

插件调用的核心思想是：当用户提出需求时，AI模型先判断是否需要调用外部工具，如果需要，就生成工具调用的指令，然后由系统执行这个指令，最后把结果返回给AI模型，由模型整理后回复给用户。

整个过程可以简化为：

1. 用户提问 → 2. AI分析需求 → 3. 选择合适工具 → 4. 执行工具 → 5. 整理结果 → 6. 回复用户

4.2 实现一个简单的天气查询插件

让我们从最简单的例子开始。假设我们要实现一个天气查询功能。

第一步：创建插件配置文件

在项目目录下创建plugins/weather.py：

import requests
import json

class WeatherPlugin:
    """天气查询插件"""
    
    def __init__(self):
        self.api_key = "your_api_key"  # 这里需要替换为真实的API密钥
        self.base_url = "https://api.weather.com/v3"
    
    def get_weather(self, city: str) -> dict:
        """获取指定城市的天气信息"""
        try:
            # 这里简化了实际的API调用
            # 实际使用时需要根据具体的天气API文档实现
            response = {
                "city": city,
                "temperature": "25°C",
                "condition": "晴朗",
                "humidity": "65%",
                "wind_speed": "3级"
            }
            return response
        except Exception as e:
            return {"error": f"获取天气信息失败: {str(e)}"}
    
    def describe(self) -> dict:
        """返回插件的描述信息"""
        return {
            "name": "weather",
            "description": "查询城市天气信息",
            "parameters": {
                "city": {
                    "type": "string",
                    "description": "城市名称，如'北京'、'上海'"
                }
            }
        }

第二步：修改Chainlit应用集成插件

修改app.py文件：

import chainlit as cl
from plugins.weather import WeatherPlugin

# 初始化插件
weather_plugin = WeatherPlugin()

@cl.on_chat_start
async def start():
    # 告诉用户可用的功能
    await cl.Message(
        content="你好！我是你的AI助手，我可以帮你查询天气、搜索信息等。试试问我'北京天气怎么样？'"
    ).send()

@cl.on_message
async def main(message: cl.Message):
    user_input = message.content
    
    # 判断是否需要调用天气插件
    if "天气" in user_input or "weather" in user_input.lower():
        # 提取城市名称（这里简化处理，实际应该用更智能的方式）
        city = extract_city(user_input)
        
        if city:
            # 调用天气插件
            weather_info = weather_plugin.get_weather(city)
            
            if "error" not in weather_info:
                response = f"{city}的天气情况：\n"
                response += f"温度：{weather_info['temperature']}\n"
                response += f"天气：{weather_info['condition']}\n"
                response += f"湿度：{weather_info['humidity']}\n"
                response += f"风速：{weather_info['wind_speed']}"
            else:
                response = f"抱歉，获取{city}的天气信息失败：{weather_info['error']}"
        else:
            response = "请告诉我你要查询哪个城市的天气，比如'北京天气怎么样？'"
    else:
        # 普通对话，调用ERNIE模型
        response = await call_ernie_model(user_input)
    
    await cl.Message(content=response).send()

def extract_city(text: str) -> str:
    """从文本中提取城市名称（简化版）"""
    # 这里应该用更智能的方法，比如NER实体识别
    # 为了简单，我们假设城市名在"天气"前面
    if "天气" in text:
        parts = text.split("天气")
        if parts[0]:
            return parts[0].strip()
    return ""

async def call_ernie_model(prompt: str) -> str:
    """调用ERNIE模型生成回复"""
    # 这里调用vLLM服务的API
    import aiohttp
    import json
    
    async with aiohttp.ClientSession() as session:
        payload = {
            "prompt": prompt,
            "max_tokens": 500,
            "temperature": 0.7
        }
        
        async with session.post(
            "http://localhost:8000/v1/completions",
            json=payload
        ) as response:
            result = await response.json()
            return result.get("choices", [{}])[0].get("text", "抱歉，我暂时无法回答这个问题。")

第三步：测试插件功能

重启Chainlit服务，然后在聊天界面输入：

北京今天天气怎么样？

你会看到AI不仅理解了你的问题，还调用了天气插件获取实时信息，然后整理成友好的格式回复给你。

4.3 扩展更多插件类型

一旦掌握了基本原理，你可以轻松添加更多插件。以下是一些常见的插件类型：

信息查询类：

• 天气预报
• 股票行情
• 汇率换算
• 航班查询

内容处理类：

• 文档总结
• 翻译服务
• 代码检查
• 格式转换

工具集成类：

• 日历管理
• 邮件发送
• 任务提醒
• 文件操作

每个插件的实现模式都类似：定义插件类 → 实现核心功能 → 集成到主应用 → 测试验证。

5. 高级功能：让AI智能选择插件

上面的实现虽然能用，但还不够智能。我们需要让AI自己判断什么时候该用什么插件。这就需要用到函数调用（Function Calling）功能。

5.1 实现智能插件路由

我们可以改造一下消息处理逻辑，让ERNIE模型帮我们决定是否需要调用插件，以及调用哪个插件。

第一步：定义插件注册表

class PluginManager:
    """插件管理器"""
    
    def __init__(self):
        self.plugins = {}
    
    def register(self, plugin):
        """注册插件"""
        info = plugin.describe()
        self.plugins[info["name"]] = {
            "instance": plugin,
            "info": info
        }
    
    def get_available_functions(self):
        """获取所有可用的函数描述"""
        functions = []
        for plugin_info in self.plugins.values():
            functions.append({
                "name": plugin_info["info"]["name"],
                "description": plugin_info["info"]["description"],
                "parameters": plugin_info["info"]["parameters"]
            })
        return functions
    
    def execute(self, function_name: str, arguments: dict):
        """执行插件函数"""
        if function_name in self.plugins:
            plugin = self.plugins[function_name]["instance"]
            # 根据函数名调用对应的方法
            if function_name == "weather":
                return plugin.get_weather(**arguments)
            # 可以添加更多插件的调用逻辑
        return None

# 初始化插件管理器
plugin_manager = PluginManager()
plugin_manager.register(WeatherPlugin())

第二步：让ERNIE模型决定是否调用插件

修改消息处理逻辑：

async def process_with_plugins(user_input: str) -> str:
    """使用插件处理用户输入"""
    
    # 第一步：让模型判断是否需要调用插件
    system_prompt = """你是一个AI助手，可以调用各种工具帮助用户。
    可用的工具：
    1. weather - 查询天气信息，参数：city（城市名称）
    
    请根据用户的问题，判断是否需要调用工具。
    如果需要，请以JSON格式回复，包含：
    - need_tool: true
    - tool_name: 工具名称
    - arguments: 参数对象
    
    如果不需要，请直接回复答案。"""
    
    # 调用ERNIE模型进行判断
    judgment = await call_ernie_model(
        f"{system_prompt}\n\n用户问题：{user_input}"
    )
    
    # 解析模型的回复
    try:
        # 尝试解析JSON
        import json
        import re
        
        # 提取JSON部分
        json_match = re.search(r'\{.*\}', judgment, re.DOTALL)
        if json_match:
            result = json.loads(json_match.group())
            if result.get("need_tool"):
                # 调用插件
                tool_result = plugin_manager.execute(
                    result["tool_name"],
                    result["arguments"]
                )
                
                # 把插件结果给模型，让模型整理回复
                final_prompt = f"""用户的问题：{user_input}
                
                工具调用结果：{json.dumps(tool_result, ensure_ascii=False)}
                
                请根据工具调用结果，用友好的方式回答用户的问题。"""
                
                return await call_ernie_model(final_prompt)
    except:
        pass
    
    # 如果不需要调用插件或解析失败，直接使用原始回复
    return judgment

这个方案的好处是：AI模型自己决定什么时候该用插件，用什么插件，参数是什么。这样更加智能和灵活。

5.2 处理复杂多轮对话

在实际使用中，用户的问题可能涉及多个插件，或者需要多轮交互才能完成。我们需要让AI记住对话历史。

class ConversationManager:
    """对话管理器，维护对话历史"""
    
    def __init__(self):
        self.history = []
    
    def add_message(self, role: str, content: str):
        """添加消息到历史"""
        self.history.append({
            "role": role,
            "content": content,
            "timestamp": time.time()
        })
        
        # 保持历史记录在合理长度
        if len(self.history) > 20:
            self.history = self.history[-20:]
    
    def get_context(self, max_turns: int = 5):
        """获取最近的对话上下文"""
        return self.history[-max_turns*2:] if len(self.history) > max_turns*2 else self.history
    
    def clear(self):
        """清空对话历史"""
        self.history = []

# 在Chainlit中使用
conversation_manager = ConversationManager()

@cl.on_message
async def handle_message(message: cl.Message):
    user_input = message.content
    
    # 添加到对话历史
    conversation_manager.add_message("user", user_input)
    
    # 获取对话上下文
    context = conversation_manager.get_context()
    
    # 构建包含上下文的提示
    context_prompt = "\n".join([f"{msg['role']}: {msg['content']}" for msg in context])
    full_prompt = f"对话历史：\n{context_prompt}\n\n用户最新消息：{user_input}\n\n请回复："
    
    # 处理消息（可能调用插件）
    response = await process_with_plugins(full_prompt)
    
    # 添加到对话历史
    conversation_manager.add_message("assistant", response)
    
    await cl.Message(content=response).send()

这样，AI就能记住之前的对话内容，实现真正的多轮对话。

6. 实战案例：构建个人AI助手

让我们用一个完整的例子来展示如何用这个平台构建一个实用的个人AI助手。

6.1 定义助手功能

假设我们要构建一个具有以下功能的助手：

1. 天气查询
2. 日程管理
3. 快速笔记
4. 信息搜索
5. 内容生成

6.2 实现核心插件

日程管理插件：

import json
import os
from datetime import datetime

class CalendarPlugin:
    """日程管理插件"""
    
    def __init__(self, storage_file="calendar.json"):
        self.storage_file = storage_file
        self.load_events()
    
    def load_events(self):
        """加载日程事件"""
        if os.path.exists(self.storage_file):
            with open(self.storage_file, 'r', encoding='utf-8') as f:
                self.events = json.load(f)
        else:
            self.events = []
    
    def save_events(self):
        """保存日程事件"""
        with open(self.storage_file, 'w', encoding='utf-8') as f:
            json.dump(self.events, f, ensure_ascii=False, indent=2)
    
    def add_event(self, title: str, date: str, time: str = "", description: str = "") -> dict:
        """添加日程事件"""
        event = {
            "id": len(self.events) + 1,
            "title": title,
            "date": date,
            "time": time,
            "description": description,
            "created_at": datetime.now().isoformat()
        }
        self.events.append(event)
        self.save_events()
        return {"success": True, "event": event}
    
    def get_events(self, date: str = None) -> dict:
        """获取日程事件"""
        if date:
            filtered = [e for e in self.events if e["date"] == date]
            return {"events": filtered, "count": len(filtered)}
        else:
            return {"events": self.events, "count": len(self.events)}
    
    def describe(self) -> dict:
        return {
            "name": "calendar",
            "description": "管理个人日程安排",
            "parameters": {
                "action": {
                    "type": "string",
                    "description": "操作类型：add（添加）、get（查询）"
                },
                "title": {
                    "type": "string",
                    "description": "事件标题（添加时必填）"
                },
                "date": {
                    "type": "string",
                    "description": "日期，格式：YYYY-MM-DD"
                }
            }
        }

笔记插件：

class NotesPlugin:
    """快速笔记插件"""
    
    def __init__(self, storage_file="notes.json"):
        self.storage_file = storage_file
        self.notes = self.load_notes()
    
    def load_notes(self):
        """加载笔记"""
        try:
            with open(self.storage_file, 'r', encoding='utf-8') as f:
                return json.load(f)
        except:
            return []
    
    def save_notes(self):
        """保存笔记"""
        with open(self.storage_file, 'w', encoding='utf-8') as f:
            json.dump(self.notes, f, ensure_ascii=False, indent=2)
    
    def add_note(self, content: str, tags: list = None) -> dict:
        """添加笔记"""
        note = {
            "id": len(self.notes) + 1,
            "content": content,
            "tags": tags or [],
            "created_at": datetime.now().isoformat()
        }
        self.notes.append(note)
        self.save_notes()
        return {"success": True, "note": note}
    
    def search_notes(self, keyword: str = None, tag: str = None) -> dict:
        """搜索笔记"""
        results = self.notes
        
        if keyword:
            results = [n for n in results if keyword in n["content"]]
        
        if tag:
            results = [n for n in results if tag in n["tags"]]
        
        return {"notes": results, "count": len(results)}
    
    def describe(self) -> dict:
        return {
            "name": "notes",
            "description": "管理个人笔记",
            "parameters": {
                "action": {
                    "type": "string", 
                    "description": "操作类型：add（添加）、search（搜索）"
                },
                "content": {
                    "type": "string",
                    "description": "笔记内容（添加时必填）"
                }
            }
        }

6.3 集成所有插件

# 初始化所有插件
plugin_manager = PluginManager()
plugin_manager.register(WeatherPlugin())
plugin_manager.register(CalendarPlugin())
plugin_manager.register(NotesPlugin())

# 更新可用的函数列表
available_functions = plugin_manager.get_available_functions()

# 更新系统提示，包含所有可用工具
system_prompt = f"""你是一个多功能AI助手，可以调用以下工具帮助用户：

{json.dumps(available_functions, ensure_ascii=False, indent=2)}

请根据用户的问题，判断是否需要调用工具，以及调用哪个工具。
如果需要调用工具，请以JSON格式回复。
如果不需要，请直接回答用户的问题。"""

6.4 测试完整功能

现在你可以测试各种功能：

用户：明天下午3点有个团队会议，帮我记一下
AI：好的，已经为您添加了日程："团队会议"，时间：明天下午3点。

用户：今天有什么安排？
AI：让我查一下...今天您有一个团队会议在下午3点。

用户：记一下：项目进度需要加快
AI：已为您添加笔记："项目进度需要加快"

用户：搜索关于"项目"的笔记
AI：找到1条相关笔记：
1. 项目进度需要加快（添加于2024-01-15）

7. 部署优化与生产建议

当你把这个平台用于实际生产环境时，还需要考虑一些优化措施。

7.1 性能优化建议

模型推理优化：

# 使用vLLM的批处理功能提高吞吐量
from vllm import SamplingParams

# 批量处理请求
sampling_params = SamplingParams(
    temperature=0.7,
    max_tokens=500,
    top_p=0.9
)

# 可以同时处理多个请求
prompts = [
    "你好，今天天气怎么样？",
    "请帮我总结一下这篇文章",
    "写一个Python函数计算斐波那契数列"
]

# 批量生成
outputs = llm.generate(prompts, sampling_params)

插件调用优化：

• 为耗时插件添加缓存机制
• 使用异步调用避免阻塞
• 设置插件调用超时时间

7.2 安全考虑

输入验证：

def validate_input(user_input: str) -> bool:
    """验证用户输入的安全性"""
    # 检查长度
    if len(user_input) > 1000:
        return False
    
    # 检查是否有危险字符
    dangerous_patterns = [
        "system(", "exec(", "eval(", "os.", "subprocess"
    ]
    
    for pattern in dangerous_patterns:
        if pattern in user_input.lower():
            return False
    
    return True

插件权限控制：

class PluginWithPermission:
    """带权限控制的插件"""
    
    def __init__(self):
        self.allowed_actions = ["read", "query"]
        self.blocked_actions = ["delete", "format", "shutdown"]
    
    def check_permission(self, action: str, user: str) -> bool:
        """检查用户是否有权限执行该操作"""
        user_roles = self.get_user_roles(user)
        
        if action in self.blocked_actions:
            return False
        
        if action in ["write", "modify"]:
            return "admin" in user_roles
        
        return True

7.3 监控与日志

添加详细的日志记录，方便排查问题：

import logging
from datetime import datetime

class Monitoring:
    """监控和日志记录"""
    
    def __init__(self):
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler('ai_platform.log'),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger(__name__)
    
    def log_request(self, user_input: str, response: str, 
                   used_plugin: str = None, duration: float = None):
        """记录请求日志"""
        log_entry = {
            "timestamp": datetime.now().isoformat(),
            "input": user_input[:100],  # 只记录前100字符
            "response_length": len(response),
            "plugin": used_plugin,
            "duration": duration
        }
        
        self.logger.info(f"Request: {log_entry}")
    
    def log_error(self, error: Exception, context: str = ""):
        """记录错误日志"""
        self.logger.error(f"Error in {context}: {str(error)}")

8. 总结与展望

通过ERNIE-4.5-0.3B-PT和Chainlit，我们成功构建了一个功能丰富的AI Agent平台。整个过程虽然涉及多个组件，但每个部分都不复杂，关键是理解整个系统的工作流程。

8.1 关键收获

技术层面：

1. 轻量模型也能做大事：0.3B的模型配合合适的架构，完全可以满足大多数应用场景
2. 插件化设计很重要：通过插件系统，我们可以轻松扩展AI的能力
3. 对话上下文是关键：维护好对话历史，AI才能理解复杂的多轮对话

实践层面：

1. 从简单开始：先实现核心功能，再逐步完善
2. 重视用户体验：好的交互设计能让技术价值最大化
3. 持续迭代优化：根据用户反馈不断改进功能和体验

8.2 下一步可以做什么

如果你对这个平台感兴趣，还可以继续扩展：

更多插件集成：

• 邮件客户端集成
• 文件管理系统
• 第三方API对接（如GitHub、Jira等）

功能增强：

• 支持语音输入输出
• 添加多轮对话记忆
• 实现个性化学习

部署优化：

• 容器化部署
• 负载均衡
• 自动扩缩容

8.3 给初学者的建议

如果你是第一次尝试构建这样的系统，我的建议是：

1. 先跑起来：不要追求完美，先让最基本的对话功能工作起来
2. 小步快跑：每次只添加一个功能，测试通过后再继续
3. 多测试：不同场景、不同输入都要测试
4. 关注用户反馈：实际使用中发现问题，比任何测试都有效

最重要的是，享受这个过程。看着自己构建的AI助手从简单对话到能调用各种工具，这种成就感是无可替代的。

---

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。