基于Phi-3-Mini-128K的智能体(Agent)开发入门:Skills框架实践

最近有不少朋友问我,现在大模型这么火,除了聊天和生成内容,能不能让它真正“动”起来,帮我们自动处理一些事情?比如,让它根据我的邮件内容自动安排日程,或者分析一份报告后自动生成摘要并保存。

答案是肯定的,这就是AI智能体(Agent)在做的事情。而要让一个模型从“能说会道”变得“能动手干活”,关键就是给它装上“技能”(Skills)。今天,我就以轻量但能力不俗的Phi-3-Mini-128K模型为例,带你从零开始,亲手搭建一个能听你指挥、帮你干活的智能体。我们会使用一个非常直观的Skills框架,整个过程就像给机器人编程一样简单有趣。

1. 开篇:为什么需要智能体与Skills?

你可能已经用过很多AI对话产品,它们能回答你的问题,甚至写诗、编程。但你会发现,它们的“能力”是固定的,只能处理文本。如果你想让AI帮你查一下明天的天气,或者把你刚写好的会议纪要保存到电脑的某个文件夹,它通常就无能为力了。

智能体(Agent) 就是为了解决这个问题而生的。你可以把它想象成一个有“大脑”和“手脚”的虚拟助手。大脑负责理解你的意图、规划和决策,而手脚就是执行具体动作的工具。这里的“手脚”,在技术层面,就是我们今天要重点讲的 Skills(技能)

一个智能体的强大与否,很大程度上取决于它拥有多少实用、可靠的Skills。通过Skills框架,我们可以轻松地为Phi-3-Mini-128K这样的大模型“安装”新能力,让它从聊天机器人升级为任务执行者。

2. 环境准备:搭建你的智能体开发沙盒

在开始写代码之前,我们需要先把“工作台”搭好。这里我推荐使用Python环境,因为它有最丰富的AI生态库。

2.1 基础环境配置

首先,确保你的电脑上安装了Python(建议3.8以上版本)。然后,我们创建一个新的项目文件夹,并安装最核心的几个库。

打开你的终端或命令行工具,依次执行以下命令:

# 1. 创建项目文件夹并进入
mkdir phi3-agent-demo
cd phi3-agent-demo

# 2. 创建并激活虚拟环境(推荐,避免包冲突)
python -m venv venv
# 在Windows上激活:
# venv\Scripts\activate
# 在Mac/Linux上激活:
# source venv/bin/activate

# 3. 安装核心库
pip install transformers torch  # Phi-3模型运行的基础
pip install sentence-transformers  # 可选,用于文本嵌入,增强理解
pip install requests  # 用于让Skill访问网络API

2.2 获取与加载Phi-3-Mini-128K模型

Phi-3-Mini是微软推出的一个非常高效的轻量级模型,128K代表它支持超长的上下文,非常适合作为智能体的“大脑”,因为它能记住很长的对话历史和复杂的任务指令。

我们使用Hugging Face的 transformers 库来加载它。由于模型文件较大(约几个GB),第一次运行时会自动下载,请确保网络通畅。

# model_loader.py
from transformers import AutoTokenizer, AutoModelForCausalLM
import torch

def load_phi3_model():
    """
    加载Phi-3-Mini-128K模型和分词器。
    """
    model_name = "microsoft/Phi-3-mini-128k-instruct"

    print("正在加载分词器...")
    tokenizer = AutoTokenizer.from_pretrained(model_name, trust_remote_code=True)

    print("正在加载模型...(首次下载需要较长时间)")
    # 根据你的显卡情况,可以选择不同的加载方式
    # 方式1:使用GPU(如果可用)
    if torch.cuda.is_available():
        model = AutoModelForCausalLM.from_pretrained(
            model_name,
            torch_dtype=torch.float16, # 半精度,节省显存
            device_map="auto",
            trust_remote_code=True
        )
        print(f"模型已加载至 GPU: {torch.cuda.get_device_name(0)}")
    # 方式2:使用CPU(如果无GPU或显存不足)
    else:
        model = AutoModelForCausalLM.from_pretrained(
            model_name,
            torch_dtype=torch.float32,
            device_map="cpu",
            trust_remote_code=True
        )
        print("模型已加载至 CPU。")

    return model, tokenizer

if __name__ == "__main__":
    model, tokenizer = load_phi3_model()
    print("模型与分词器加载完毕!")

运行这个脚本,如果看到成功加载的信息,那么你的“大脑”就准备好了。

3. Skills框架核心:如何让AI学会“动手”

Skills框架的核心思想是将外部能力封装成统一的、模型可以理解和调用的函数。一个典型的Skill包含三部分:描述、参数和函数体。

3.1 设计你的第一个Skill:获取时间

让我们从一个最简单的Skill开始,让智能体能够获取当前时间。这虽然简单,但能让你立刻理解整个流程。

# skills/basic_skills.py
import json
from datetime import datetime

class SkillBase:
    """所有Skill的基类,定义通用接口。"""
    def __init__(self, name, description, parameters):
        self.name = name
        self.description = description
        self.parameters = parameters  # 参数格式说明

    def execute(self, **kwargs):
        """执行技能的具体逻辑,由子类实现。"""
        raise NotImplementedError

class GetCurrentTimeSkill(SkillBase):
    """获取当前日期和时间的技能。"""
    def __init__(self):
        # 用自然语言清晰描述这个技能是干什么的
        description = "获取当前的日期和时间。当用户询问时间、日期或现在几点时使用。"
        # 定义这个技能需要什么参数(这个技能不需要额外参数)
        parameters = {
            "type": "object",
            "properties": {},
            "required": []
        }
        super().__init__(name="get_current_time", description=description, parameters=parameters)

    def execute(self, **kwargs):
        """执行函数:返回当前时间字符串。"""
        now = datetime.now()
        # 返回一个结构化的结果,方便模型理解
        return {
            "status": "success",
            "data": now.strftime("%Y年%m月%d日 %H:%M:%S"),
            "message": f"当前时间是:{now.strftime('%Y年%m月%d日 %H:%M:%S')}"
        }

# 技能管理器:注册和管理所有可用技能
class SkillManager:
    def __init__(self):
        self.skills = {}

    def register_skill(self, skill: SkillBase):
        """注册一个技能。"""
        self.skills[skill.name] = skill
        print(f"技能已注册: {skill.name}")

    def get_skill(self, skill_name):
        """根据名称获取技能。"""
        return self.skills.get(skill_name)

    def list_skills_for_model(self):
        """生成供模型理解的技能描述列表。"""
        skill_list = []
        for name, skill in self.skills.items():
            skill_list.append({
                "name": name,
                "description": skill.description,
                "parameters": skill.parameters
            })
        return skill_list

# 初始化技能管理器并注册技能
skill_manager = SkillManager()
skill_manager.register_skill(GetCurrentTimeSkill())

3.2 进阶Skill:网络搜索与文件操作

一个只会报时的助手显然不够看。让我们为它添加更实用的技能:搜索网页和创建文件。

# skills/advanced_skills.py
import requests
import os
from .basic_skills import SkillBase, skill_manager

class WebSearchSkill(SkillBase):
    """一个简单的网络搜索技能(使用模拟API或公共API)。"""
    def __init__(self):
        # 注意:这里使用一个模拟的搜索端点。在实际应用中,你需要替换为真实的搜索引擎API(如Serper、Google Custom Search等)。
        description = "在互联网上搜索信息。当用户需要查找最新资讯、事实核查或未知信息时使用。"
        parameters = {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "需要搜索的关键词或问题。"
                }
            },
            "required": ["query"]
        }
        super().__init__(name="web_search", description=description, parameters=parameters)

    def execute(self, query):
        """执行搜索。"""
        print(f"[技能执行] 正在搜索: {query}")
        # 这里是模拟搜索,实际使用时请接入真正的API
        # 例如,使用 requests.get(f"https://api.serper.dev/search?q={query}")
        mock_results = [
            f"关于'{query}'的模拟结果1:这是相关的信息摘要。",
            f"关于'{query}'的模拟结果2:根据网络资料,这是另一条信息。"
        ]
        return {
            "status": "success",
            "data": mock_results,
            "message": f"已找到{len(mock_results)}条关于'{query}'的信息。"
        }

class CreateNoteSkill(SkillBase):
    """在指定路径创建文本文件的技能。"""
    def __init__(self, base_path="./notes"):
        description = "在本地创建一个文本文件(笔记)。当用户要求保存信息、记录想法或创建备忘录时使用。"
        parameters = {
            "type": "object",
            "properties": {
                "filename": {
                    "type": "string",
                    "description": "要创建的文件名(无需后缀,会自动添加.txt)。"
                },
                "content": {
                    "type": "string",
                    "description": "要写入文件的内容。"
                }
            },
            "required": ["filename", "content"]
        }
        self.base_path = base_path
        os.makedirs(self.base_path, exist_ok=True)
        super().__init__(name="create_note", description=description, parameters=parameters)

    def execute(self, filename, content):
        """创建并写入文件。"""
        # 确保文件名以.txt结尾
        if not filename.endswith('.txt'):
            filename += '.txt'
        filepath = os.path.join(self.base_path, filename)

        try:
            with open(filepath, 'w', encoding='utf-8') as f:
                f.write(content)
            return {
                "status": "success",
                "data": {"filepath": filepath},
                "message": f"笔记已成功创建并保存至:{filepath}"
            }
        except Exception as e:
            return {
                "status": "error",
                "data": None,
                "message": f"创建文件失败:{str(e)}"
            }

# 注册进阶技能
skill_manager.register_skill(WebSearchSkill())
skill_manager.register_skill(CreateNoteSkill())

现在,你的智能体已经拥有了三个技能:报时、搜索和记笔记。技能库可以像这样不断扩展,比如添加发送邮件、查询数据库、控制智能家居等。

4. 智能体大脑:连接模型与Skills

有了技能,下一步就是让Phi-3模型学会在合适的时机调用它们。这个过程称为任务规划与执行。我们需要做两件事:

  1. 让模型知道有哪些技能可用:将技能描述格式化后,放入模型的系统提示词中。
  2. 让模型学会输出调用指令:引导模型以特定的格式(如JSON)来“思考”和“决定”调用哪个技能。

4.1 构建系统提示词

系统提示词是告诉模型“你是谁”、“你能做什么”的关键。我们将可用的技能列表格式化后嵌入其中。

# agent/core.py
import re
import json
from skills.basic_skills import skill_manager

class Phi3Agent:
    def __init__(self, model, tokenizer):
        self.model = model
        self.tokenizer = tokenizer
        self.skill_manager = skill_manager
        self.conversation_history = [] # 记录对话历史,用于长上下文

    def _build_system_prompt(self):
        """构建包含可用技能描述的系统提示词。"""
        skills_info = self.skill_manager.list_skills_for_model()
        skills_json_str = json.dumps(skills_info, indent=2, ensure_ascii=False)

        system_prompt = f"""你是一个AI智能体,可以调用以下工具(Skills)来帮助用户完成任务:
{skills_json_str}

**调用规则**:
1.  当你判断需要调用工具时,请严格按照以下JSON格式输出,且只输出这个JSON,不要有任何其他文字:
```json
{{"action": "call_skill", "skill_name": "技能名称", "arguments": {{"参数名": "参数值"}}}}
  1. 如果用户的问题不需要调用工具,或者只是普通聊天,请直接以自然语言回复。
  2. 一次只调用一个工具。

请根据用户请求,决定是否需要调用工具,以及调用哪个工具。 """ return system_prompt

def _parse_model_response(self, response_text):
    """
    解析模型的回复,判断是直接回答还是调用了技能。
    尝试从回复中提取JSON格式的调用指令。
    """
    # 尝试查找JSON块
    json_pattern = r'```json\s*(.*?)\s*```'
    match = re.search(json_pattern, response_text, re.DOTALL)
    if match:
        json_str = match.group(1)
        try:
            action_data = json.loads(json_str)
            if action_data.get("action") == "call_skill":
                return action_data
        except json.JSONDecodeError:
            pass

    # 如果没有找到标准的JSON代码块,尝试直接解析整个回复是否为JSON(有些模型可能不输出```)
    try:
        action_data = json.loads(response_text.strip())
        if action_data.get("action") == "call_skill":
            return action_data
    except json.JSONDecodeError:
        pass

    # 如果都不是,则认为是自然语言回复
    return {"action": "direct_reply", "content": response_text}
### 4.2 实现对话与执行循环

这是智能体的主循环,负责与用户交互、调用模型、解析意图并执行技能。

```python
# agent/core.py (续)
class Phi3Agent:
    # ... __init__ 和 _build_system_prompt 方法 ...

    def chat_round(self, user_input):
        """处理用户的一轮输入。"""
        # 1. 构建本次对话的完整提示
        system_prompt = self._build_system_prompt()
        history_prompt = "\n".join([f"User: {h['user']}\nAssistant: {h['assistant']}" for h in self.conversation_history[-5:]]) # 保留最近5轮历史
        full_prompt = f"{system_prompt}\n\n对话历史:\n{history_prompt}\n\n当前用户输入:{user_input}\n\nAssistant:"

        # 2. 让模型生成回复
        inputs = self.tokenizer(full_prompt, return_tensors="pt", truncation=True, max_length=4096).to(self.model.device)
        with torch.no_grad():
            outputs = self.model.generate(**inputs, max_new_tokens=256, temperature=0.7)
        model_raw_response = self.tokenizer.decode(outputs[0][inputs['input_ids'].shape[1]:], skip_special_tokens=True)

        # 3. 解析模型回复
        parsed_action = self._parse_model_response(model_raw_response)

        final_response = ""
        if parsed_action["action"] == "call_skill":
            # 执行技能调用
            skill_name = parsed_action["skill_name"]
            arguments = parsed_action.get("arguments", {})
            skill = self.skill_manager.get_skill(skill_name)

            if skill:
                try:
                    result = skill.execute(**arguments)
                    final_response = f"【已执行技能 `{skill_name}`】\n结果:{result['message']}"
                except Exception as e:
                    final_response = f"执行技能 `{skill_name}` 时出错:{str(e)}"
            else:
                final_response = f"未知技能:{skill_name}"
        else:
            # 直接回复
            final_response = parsed_action["content"]

        # 4. 更新对话历史
        self.conversation_history.append({"user": user_input, "assistant": final_response})
        # 保持历史记录不会无限增长
        if len(self.conversation_history) > 10:
            self.conversation_history.pop(0)

        return final_response

    def interactive_chat(self):
        """启动一个交互式聊天循环。"""
        print("Phi-3智能体已启动!输入 '退出' 或 'quit' 结束对话。")
        print(f"当前已加载技能:{list(self.skill_manager.skills.keys())}")
        while True:
            try:
                user_input = input("\n你:")
                if user_input.lower() in ['退出', 'quit', 'exit']:
                    print("对话结束。")
                    break
                response = self.chat_round(user_input)
                print(f"\n助手:{response}")
            except KeyboardInterrupt:
                print("\n对话被中断。")
                break
            except Exception as e:
                print(f"\n发生错误:{e}")

5. 实战:打造你的日程管理智能体

现在,让我们把前面所有的部分组合起来,创建一个简单的日程管理智能体。它会结合“获取时间”和“创建笔记”技能,实现一个经典场景:根据用户描述,创建日程提醒文件

我们需要新增一个更专业的Skill,并优化提示词。

5.1 创建日程管理Skill

这个Skill比简单的创建笔记更智能一些,它会按照日期自动组织文件。

# skills/schedule_skill.py
import os
from datetime import datetime
from .basic_skills import SkillBase, skill_manager

class CreateScheduleSkill(SkillBase):
    """创建日程安排文件的技能。"""
    def __init__(self, base_path="./schedules"):
        description = "为用户创建或添加日程安排。根据用户提供的日期、时间和事件描述,生成一个结构化的日程文件。如果未提供日期,则使用今天。"
        parameters = {
            "type": "object",
            "properties": {
                "date": {
                    "type": "string",
                    "description": "日程的日期,格式为'YYYY-MM-DD',例如'2024-05-20'。如果用户说'今天'、'明天',你需要将其转换为具体日期。"
                },
                "time": {
                    "type": "string",
                    "description": "日程的具体时间,例如'14:30'、'上午10点'。"
                },
                "event": {
                    "type": "string",
                    "description": "日程事件的详细描述。"
                }
            },
            "required": ["event"] # 只有事件描述是必须的
        }
        self.base_path = base_path
        os.makedirs(self.base_path, exist_ok=True)
        super().__init__(name="create_schedule", description=description, parameters=parameters)

    def execute(self, event, date=None, time=None):
        """创建或更新日程文件。"""
        # 处理日期:如果未提供,默认为今天
        if not date:
            date = datetime.now().strftime("%Y-%m-%d")
        # 简单处理一下“明天”之类的相对日期(这里做简单演示,实际需要更复杂的NLP解析)
        if date == "明天":
            from datetime import timedelta
            date = (datetime.now() + timedelta(days=1)).strftime("%Y-%m-%d")
        elif date == "今天":
            date = datetime.now().strftime("%Y-%m-%d")

        filename = f"schedule_{date}.txt"
        filepath = os.path.join(self.base_path, filename)

        # 构建日程内容
        schedule_content = f"日期:{date}\n"
        if time:
            schedule_content += f"时间:{time}\n"
        schedule_content += f"事件:{event}\n"
        schedule_content += "-" * 20 + "\n"

        # 如果文件已存在,则追加内容;否则创建新文件
        mode = 'a' if os.path.exists(filepath) else 'w'
        try:
            with open(filepath, mode, encoding='utf-8') as f:
                f.write(schedule_content)
            action = "已添加到" if mode == 'a' else "已创建"
            return {
                "status": "success",
                "data": {"filepath": filepath, "date": date},
                "message": f"日程{action}文件:{filepath}"
            }
        except Exception as e:
            return {
                "status": "error",
                "data": None,
                "message": f"操作日程文件失败:{str(e)}"
            }

# 注册新技能
skill_manager.register_skill(CreateScheduleSkill())

5.2 运行你的智能体

创建一个主程序文件,将所有模块串联起来。

# main.py
from model_loader import load_phi3_model
from agent.core import Phi3Agent
# 导入技能模块以完成注册
import skills.basic_skills
import skills.advanced_skills
import skills.schedule_skill

def main():
    print("=== 启动 Phi-3-Mini 智能体 ===")
    # 1. 加载模型
    model, tokenizer = load_phi3_model()

    # 2. 创建智能体
    agent = Phi3Agent(model, tokenizer)

    # 3. 启动交互式对话
    agent.interactive_chat()

if __name__ == "__main__":
    main()

现在,运行 python main.py,你的智能体就启动了!试试和它对话:

你:现在几点了?
助手:【已执行技能 `get_current_time`】
结果:当前时间是:2024年05月20日 14:25:33

你:帮我记一下,明天下午3点要和客户开项目评审会。
助手:【已执行技能 `create_schedule`】
结果:日程已添加到文件:./schedules/schedule_2024-05-21.txt

你:搜索一下今天的热点新闻。
助手:【已执行技能 `web_search`】
结果:已找到2条关于'今天的热点新闻'的信息。

看,它已经能理解你的意图,并调用不同的技能来完成任务了!

6. 总结

跟着上面的步骤走一遍,你应该已经成功搭建了一个基于Phi-3-Mini和Skills框架的、能够执行具体任务的智能体原型。整个过程就像搭积木:准备一个聪明的大脑(Phi-3模型),然后为它设计各种好用的工具(Skills),最后通过一套规则(提示词与解析逻辑)让大脑学会在何时使用何种工具。

这个简单的例子展示了智能体开发的核心闭环:理解 -> 规划 -> 执行 -> 反馈。目前我们的智能体还比较基础,比如技能调用是手写规则解析的,任务规划也相对简单。但这就是起点,你可以在此基础上做很多有趣的扩展:

  • 增加更多技能:比如连接日历API、发送邮件、控制智能设备,它的能力边界只取决于你的想象力。
  • 优化规划能力:引入更专业的Agent框架(如LangChain、AutoGen),让模型能处理多步骤的复杂任务,比如“查一下天气,如果下雨就提醒我带伞,并把提醒记到笔记里”。
  • 增强理解与可靠性:处理模型输出时,可以加入验证和重试机制,确保技能调用的准确性。

用Phi-3-Mini这类轻量模型来入门智能体开发非常合适,它速度快、资源消耗相对少,能让你快速验证想法。Skills框架的设计思路也极具通用性,你可以把这套方法迁移到其他模型上。最重要的是,你亲手让一段代码“活”了过来,它能听懂你的话,并为你做事。这种成就感,正是技术最迷人的地方。不妨就从今天创建的这个小助手开始,尝试给它添加第一个属于你自己的独特技能吧。


获取更多AI镜像

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

Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐