别再盲目学 AI Agent 了!微软官方入门教程让我彻底搞懂了智能体开发全流程

别再盲目学 AI Agent 了!微软官方入门教程让我彻底搞懂了智能体开发全流程

别再盲目学 AI Agent 了!微软官方入门教程让我彻底搞懂了智能体开发全流程


为什么这个项目值得关注

在人工智能飞速发展的今天,AI Agent(人工智能智能体)已经成为了最热门的技术方向之一。从 OpenAI 的 GPT-4 到各大科技公司的智能助手,Agent 概念正在重塑我们与计算机交互的方式。然而,对于刚刚踏入这个领域的学习者来说,市面上的资料要么过于理论化、缺乏实战内容,要么就是门槛太高、让人望而却步。

微软开源的 microsoft/ai-agents-for-beginners 正是为解决这一痛点而生。这个项目采用”理论结合实践”的方式,用最通俗易懂的语言讲解 AI Agent 的核心概念,同时提供了大量可以直接运行的代码示例。无论你是计算机专业的大学生、想要转型的软件工程师,还是对 AI 充满好奇的普通爱好者,这个项目都能让你在最短的时间内掌握 AI Agent 开发的核心技能。

更重要的是,作为微软官方出品的教程,这个项目代表了一种高质量的学习路径。它不是简单地堆砌概念,而是从实际应用场景出发,让你在动手实践中理解每一个知识点。项目的代码质量有保障,文档清晰详尽,社区活跃度高,遇到问题能够及时获得帮助。


环境搭建:从零开始配置开发环境

系统要求与前置准备

在开始学习之前,我们需要确保开发环境满足基本要求。这个项目主要使用 Python 进行开发,因此你需要准备以下内容:

首先,确保你的电脑上安装了 Python 3.8 或更高版本。你可以通过访问 Python 官方网站下载最新版本的安装包。安装过程中记得勾选”Add Python to PATH”选项,这样可以在命令行中直接使用 python 命令。

其次,你需要安装 Git 版本控制工具。Git 不仅用于下载项目代码,更是现代软件开发的必备技能。你可以从 Git 官网下载安装包,安装完成后在终端中输入 git --version 验证安装是否成功。

创建虚拟环境

为了避免不同项目之间的依赖冲突,强烈建议为这个项目创建独立的虚拟环境。打开终端或命令提示符,执行以下命令创建并激活虚拟环境:

# 创建虚拟环境
python -m venv agent-env

# 在 Windows 系统激活虚拟环境
agent-env\Scripts\activate

# 在 macOS 或 Linux 系统激活虚拟环境
source agent-env/bin/activate

激活虚拟环境后,终端提示符前面会显示环境名称,表示你已经进入了这个独立的开发环境。在这个环境中安装的任何包都不会影响系统全局的 Python 环境,也不会有其他项目干扰你的学习。

安装核心依赖

这个项目的依赖管理采用 Poetry 工具,相比 pip 更加现代化,能够自动处理依赖之间的版本冲突。首先安装 Poetry:

# 使用 pip 安装 Poetry
pip install poetry

# 验证 Poetry 安装
poetry --version

接下来,克隆项目仓库并安装依赖:

# 克隆项目仓库
git clone https://github.com/microsoft/ai-agents-for-beginners.git

# 进入项目目录
cd ai-agents-for-beginners

# 安装项目依赖
poetry install

# 安装开发依赖(包括测试工具)
poetry install --with dev

如果安装过程中遇到网络问题,可以使用国内镜像源加速下载。在项目根目录下创建 poetry.toml 文件,添加以下内容:

# 配置国内镜像源
[[tool.poetry.source]]
name = "aliyun"
url = "https://mirrors.aliyun.com/pypi/simple/"
default = true

开发工具推荐

对于这个项目的学习,推荐使用 Visual Studio Code 作为代码编辑器。它不仅免费开源,而且对 Python 有良好的支持。你需要安装以下扩展来提升开发体验:

Python 扩展由微软官方提供,提供了代码补全、语法检查、调试等功能。Pylance 扩展则进一步增强了类型检查和代码导航能力。Jupyter 扩展让你能够在 VSCode 中直接运行 Jupyter Notebook,非常适合边学边练。

安装完扩展后,在 VSCode 中打开项目文件夹,按下 Ctrl+Shift+P(macOS 为 Cmd+Shift+P)打开命令面板,输入”Python: Select Interpreter”选择你创建的虚拟环境中的 Python 解释器。


核心概念详解:深入理解 AI Agent

什么是 AI Agent

在深入代码之前,我们需要理解 AI Agent 的本质概念。简单来说,AI Agent 是一个能够感知环境、做出决策并执行动作的智能系统。与传统的固定程序不同,Agent 具有一定程度的自主性和适应性,能够根据反馈调整自己的行为。

一个典型的 AI Agent 由四个核心组件构成:感知器(Perception)负责接收外部信息,可能是用户的文字输入、图像、音频或者其他数据;知识库(Knowledge)存储了 Agent 的记忆和学到的信息,类似于人类的长期记忆;推理引擎(Reasoning Engine)是 Agent 的大脑,负责分析输入、规划行动;执行器(Actuator)则负责将决策转化为具体的动作,如生成回复、操作文件、调用 API 等。

微软的这个教程将这些抽象概念具体化为可操作的代码模块,让你可以直观地看到每个组件是如何实现的。

Agent 的工作循环

AI Agent 的核心是一个持续运行的工作循环,理解这个循环是掌握 Agent 开发的关键。工作循环包含四个主要阶段:思考(Think)、规划(Plan)、执行(Act)、观察(Observe)。

在思考阶段,Agent 分析当前的状态和收到的输入,结合知识库中的信息,形成对问题的理解。规划阶段则决定下一步应该做什么,可能涉及将复杂任务分解为多个子任务,或者选择合适的工具来解决问题。执行阶段真正实施规划的行动,可能是调用外部 API、查询数据库或者生成文本回复。最后的观察阶段收集行动的结果,这些结果会成为下一轮循环的输入。

这个循环会持续进行,直到 Agent 判断任务已经完成或者达到了某个终止条件。理解了这个循环之后,你就能更容易地理解教程中各个代码模块的作用和它们之间的协作关系。

记忆系统的奥秘

记忆系统是让 AI Agent 变得”智能”的关键技术之一。没有记忆的 Agent 就像金鱼一样,每次对话都是全新的开始,无法利用之前的信息来提供更好的服务。

这个教程详细讲解了两种主要的记忆实现方式。第一种是短期记忆,也称为工作记忆,它存储当前对话上下文中的重要信息,通常使用滑动窗口的方式来管理,只保留最近的几轮对话。第二种是长期记忆,它能够跨对话保存信息,可能存储在向量数据库中,通过语义相似度来检索相关内容。

教程中的示例代码展示了如何实现一个简单的记忆管理系统,包括如何添加新的记忆、如何检索相关的记忆、以及如何在空间不足时智能地删除不重要的记忆。通过这些代码,你可以亲手实验不同记忆策略的效果。

工具调用机制

工具调用是 AI Agent 与外部世界交互的主要方式。一个强大的 Agent 之所以有用,很大程度上是因为它能够调用各种工具来完成复杂的任务。想象一下,如果一个 Agent 只能回答问题而不能执行任何操作,它的实用性将大打折扣。

在这个教程中,你会学到如何定义工具、如何让 Agent 理解工具的功能、以及如何让 Agent 智能地选择使用哪个工具。教程提供了多个实际可用的工具示例,包括网络搜索工具、计算器工具、文件读写工具等。通过这些示例,你可以掌握工具调用的核心模式,并学会如何开发自己的工具。


实战教程:一步步构建你的第一个 AI Agent

第一步:创建 Agent 的基础结构

让我们从最基础的部分开始,逐步构建一个完整的 AI Agent。首先创建一个新的 Python 文件,命名为 my_first_agent.py,然后添加以下代码:

"""
第一个 AI Agent 示例
这个示例展示了 AI Agent 的基本结构和核心组件
"""

# 导入必要的模块
import os
from typing import List, Dict, Any

# 定义消息类型,便于区分用户消息和系统消息
class Message:
    def __init__(self, role: str, content: str):
        """
        初始化消息对象

        Args:
            role: 消息角色,可以是 "user"、"assistant" 或 "system"
            content: 消息的具体内容
        """
        self.role = role
        self.content = content

    def to_dict(self) -> Dict[str, str]:
        """将消息转换为字典格式,便于与语言模型交互"""
        return {"role": self.role, "content": self.content}

class SimpleAgent:
    """一个简单但功能完整的 AI Agent"""

    def __init__(self, name: str, system_prompt: str):
        """
        初始化 Agent 实例

        Args:
            name: Agent 的名称
            system_prompt: 系统提示词,定义 Agent 的角色和行为
        """
        self.name = name
        self.system_prompt = system_prompt
        self.conversation_history: List[Message] = []

        # 初始化系统提示
        self.conversation_history.append(
            Message("system", system_prompt)
        )

    def add_message(self, role: str, content: str) -> None:
        """
        向对话历史中添加新消息

        Args:
            role: 消息角色
            content: 消息内容
        """
        message = Message(role, content)
        self.conversation_history.append(message)

    def get_conversation_history(self) -> List[Dict[str, str]]:
        """获取完整的对话历史"""
        return [msg.to_dict() for msg in self.conversation_history]

    def clear_history(self) -> None:
        """清除对话历史,但保留系统提示"""
        # 保留系统提示,删除其他所有消息
        self.conversation_history = [
            self.conversation_history[0]
        ]

这个基础结构定义了一个 Agent 类,包含初始化方法、消息添加方法和对话历史管理方法。现在这个 Agent 还不能真正与语言模型交互,我们需要继续添加推理功能。

第二步:集成语言模型

要让 Agent 能够思考和对话,我们需要集成一个语言模型。教程推荐使用 Azure OpenAI Service 或者 OpenAI API。下面我们添加模型调用功能:

# 继续在 my_first_agent.py 中添加以下代码

from openai import AzureOpenAI
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

class LLMAgent(SimpleAgent):
    """集成了语言模型的增强版 Agent"""

    def __init__(self, name: str, system_prompt: str, model: str = "gpt-4"):
        """
        初始化带语言模型的 Agent

        Args:
            name: Agent 名称
            system_prompt: 系统提示词
            model: 要使用的模型名称
        """
        super().__init__(name, system_prompt)
        self.model = model
        self._setup_client()

    def _setup_client(self) -> None:
        """设置 OpenAI 客户端"""
        # 根据环境变量判断使用 Azure OpenAI 还是 OpenAI
        if os.getenv("AZURE_OPENAI_API_KEY"):
            # Azure OpenAI 配置
            self.client = AzureOpenAI(
                api_key=os.getenv("AZURE_OPENAI_API_KEY"),
                api_version=os.getenv("AZURE_OPENAI_API_VERSION", "2024-02-01"),
                azure_endpoint=os.getenv("AZURE_OPENAI_ENDPOINT")
            )
            self.deployment = os.getenv("AZURE_OPENAI_DEPLOYMENT_NAME")
        else:
            # 标准 OpenAI 配置
            self.client = OpenAI(
                api_key=os.getenv("OPENAI_API_KEY")
            )
            self.deployment = self.model

    def think(self, user_input: str) -> str:
        """
        核心推理方法:处理用户输入并生成回复

        Args:
            user_input: 用户的输入文本

        Returns:
            Agent 的回复文本
        """
        # 将用户消息添加到历史记录
        self.add_message("user", user_input)

        # 调用语言模型
        try:
            response = self._call_llm()
            # 将助手回复添加到历史记录
            self.add_message("assistant", response)
            return response
        except Exception as e:
            error_msg = f"生成回复时出错: {str(e)}"
            print(error_msg)
            return error_msg

    def _call_llm(self) -> str:
        """调用语言模型API"""
        messages = self.get_conversation_history()

        if hasattr(self, 'client') and self.deployment:
            # 实际调用 API
            if os.getenv("AZURE_OPENAI_API_KEY"):
                response = self.client.chat.completions.create(
                    model=self.deployment,
                    messages=messages,
                    temperature=0.7,
                    max_tokens=2000
                )
            else:
                response = self.client.chat.completions.create(
                    model=self.deployment,
                    messages=messages,
                    temperature=0.7,
                    max_tokens=2000
                )
            return response.choices[0].message.content

        # 如果没有配置 API,返回模拟回复用于测试
        return "请配置您的 API 密钥以启用真正的 AI 对话功能"

现在你已经拥有了一个可以真正对话的 Agent。但这只是开始,接下来我们要添加更强大的功能。

第三步:实现记忆系统

为了让 Agent 记住之前的对话内容,我们需要实现记忆系统。下面的代码展示了如何添加短期记忆功能:

# 继续添加记忆系统实现

from datetime import datetime
from typing import Optional
import json

class Memory:
    """记忆管理模块"""

    def __init__(self, max_short_term: int = 10):
        """
        初始化记忆系统

        Args:
            max_short_term: 短期记忆保存的最大消息数
        """
        self.short_term: List[Dict[str, Any]] = []
        self.long_term: List[Dict[str, Any]] = []
        self.max_short_term = max_short_term

    def add_to_short_term(self, content: str, importance: int = 5) -> None:
        """
        添加短期记忆

        Args:
            content: 要记忆的内容
            importance: 重要性评分(1-10)
        """
        memory_entry = {
            "content": content,
            "importance": importance,
            "timestamp": datetime.now().isoformat()
        }
        self.short_term.append(memory_entry)

        # 超过容量限制时,移除不重要的记忆
        self._prune_short_term()

    def _prune_short_term(self) -> None:
        """清理不重要或过旧的短期记忆"""
        if len(self.short_term) <= self.max_short_term:
            return

        # 按重要性排序,保留最重要的记忆
        self.short_term.sort(key=lambda x: x["importance"], reverse=True)
        self.short_term = self.short_term[:self.max_short_term]

    def add_to_long_term(self, content: str, tags: List[str] = None) -> None:
        """
        添加长期记忆

        Args:
            content: 要记忆的内容
            tags: 用于检索的标签
        """
        memory_entry = {
            "content": content,
            "tags": tags or [],
            "created_at": datetime.now().isoformat(),
            "access_count": 0,
            "last_accessed": datetime.now().isoformat()
        }
        self.long_term.append(memory_entry)

    def search_long_term(self, query: str) -> Optional[List[str]]:
        """
        在长期记忆中搜索相关内容

        Args:
            query: 搜索关键词

        Returns:
            匹配的记忆内容列表
        """
        results = []
        for memory in self.long_term:
            # 简单的关键词匹配
            if query.lower() in memory["content"].lower():
                # 更新访问统计
                memory["access_count"] += 1
                memory["last_accessed"] = datetime.now().isoformat()
                results.append(memory["content"])

        return results if results else None

    def get_context_summary(self) -> str:
        """
        生成记忆上下文摘要,供 Agent 使用

        Returns:
            格式化的记忆摘要字符串
        """
        context_parts = []

        if self.short_term:
            short_summary = "最近的重要信息:"
            for mem in self.short_term[-3:]:  # 只取最近3条
                short_summary += f"\n- {mem['content']}"
            context_parts.append(short_summary)

        if self.long_term:
            long_summary = "\n保存的长期信息:"
            for mem in self.long_term[:5]:  # 取最重要的5条
                long_summary += f"\n- {mem['content']} (访问次数: {mem['access_count']})"
            context_parts.append(long_summary)

        return "\n".join(context_parts) if context_parts else "暂无记忆记录"

class AgentWithMemory(LLMAgent):
    """带有记忆功能的增强版 Agent"""

    def __init__(self, name: str, system_prompt: str, model: str = "gpt-4"):
        """初始化带记忆的 Agent"""
        super().__init__(name, system_prompt, model)
        self.memory = Memory(max_short_term=10)

    def think(self, user_input: str) -> str:
        """
        带记忆功能的思考方法

        Args:
            user_input: 用户输入

        Returns:
            Agent 的回复
        """
        # 检索相关记忆
        relevant_memory = self.memory.search_long_term(user_input)
        memory_context = self.memory.get_context_summary()

        # 将记忆融入系统提示
        enhanced_system = self.system_prompt + f"\n\n当前记忆上下文:\n{memory_context}"

        if relevant_memory:
            enhanced_system += f"\n\n相关长期记忆:\n" + "\n".join(f"- {m}" for m in relevant_memory)

        # 更新 Agent 的系统提示
        self.conversation_history[0] = Message("system", enhanced_system)

        # 处理用户输入并生成回复
        response = super().think(user_input)

        # 保存重要信息到记忆
        self.memory.add_to_short_term(f"用户问: {user_input}", importance=7)
        self.memory.add_to_short_term(f"Agent 答: {response[:100]}...", importance=5)

        return response

这个记忆系统让 Agent 能够”记住”对话中的重要信息,从而在后续对话中提供更连贯、更个性化的服务。

第四步:添加工具调用能力

工具调用是 AI Agent 最强大的特性之一。下面的代码展示了如何实现一个可扩展的工具系统:

# 添加工具系统

from abc import ABC, abstractmethod
import json
from datetime import datetime

class BaseTool(ABC):
    """所有工具的基类"""

    @property
    @abstractmethod
    def name(self) -> str:
        """工具名称"""
        pass

    @property
    @abstractmethod
    def description(self) -> str:
        """工具功能描述"""
        pass

    @property
    def parameters(self) -> Dict[str, Any]:
        """工具参数定义(JSON Schema 格式)"""
        return {}

    @abstractmethod
    def execute(self, **kwargs) -> str:
        """
        执行工具的核心逻辑

        Returns:
            执行结果的文本描述
        """
        pass

class CalculatorTool(BaseTool):
    """计算器工具"""

    @property
    def name(self) -> str:
        return "calculator"

    @property
    def description(self) -> str:
        return "执行数学计算。支持加减乘除和括号运算符。例如:2+3*4"

    @property
    def parameters(self) -> Dict[str, Any]:
        return {
            "type": "object",
            "properties": {
                "expression": {
                    "type": "string",
                    "description": "要计算的数学表达式"
                }
            },
            "required": ["expression"]
        }

    def execute(self, expression: str) -> str:
        """执行计算并返回结果"""
        try:
            # 安全计算:只允许数字和运算符
            allowed_chars = set("0123456789+-*/.() ")
            if not all(c in allowed_chars for c in expression):
                return "错误:表达式包含无效字符"

            # 使用 eval 进行计算
            result = eval(expression)
            return f"计算结果:{expression} = {result}"
        except ZeroDivisionError:
            return "错误:除数不能为零"
        except Exception as e:
            return f"计算错误:{str(e)}"

class DateTimeTool(BaseTool):
    """日期时间工具"""

    @property
    def name(self) -> str:
        return "get_current_datetime"

    @property
    def description(self) -> str:
        return "获取当前的日期和时间信息"

    @property
    def parameters(self) -> Dict[str, Any]:
        return {
            "type": "object",
            "properties": {},
            "required": []
        }

    def execute(self) -> str:
        """返回当前日期时间"""
        now = datetime.now()
        return f"当前时间:{now.strftime('%Y年%m月%d日 %H:%M:%S')}\n星期{['一','二','三','四','五','六','日'][now.weekday()]}"

class TodoTool(BaseTool):
    """待办事项工具"""

    @property
    def name(self) -> str:
        return "todo_list"

    @property
    def description(self) -> str:
        return "管理待办事项列表。可以添加、查看、标记完成或删除待办事项"

    @property
    def parameters(self) -> Dict[str, Any]:
        return {
            "type": "object",
            "properties": {
                "action": {
                    "type": "string",
                    "enum": ["add", "list", "complete", "delete"],
                    "description": "要执行的操作"
                },
                "item": {
                    "type": "string",
                    "description": "待办事项内容(添加时需要)"
                },
                "index": {
                    "type": "integer",
                    "description": "待办事项的索引(删除或完成时需要)"
                }
            },
            "required": ["action"]
        }

    def __init__(self):
        """初始化待办事项存储"""
        self.items: List[Dict[str, Any]] = []

    def execute(self, action: str, item: str = None, index: int = None) -> str:
        """执行待办事项操作"""
        if action == "add":
            if not item:
                return "错误:添加待办事项需要提供内容"
            self.items.append({"text": item, "completed": False, "created_at": datetime.now().isoformat()})
            return f"已添加待办事项:{item}"

        elif action == "list":
            if not self.items:
                return "待办事项列表为空"
            result = "当前待办事项:\n"
            for i, item in enumerate(self.items, 1):
                status = "✓" if item["completed"] else "○"
                result += f"{i}. {status} {item['text']}\n"
            return result

        elif action == "complete":
            if index is None or index < 1 or index > len(self.items):
                return "错误:无效的索引"
            self.items[index - 1]["completed"] = True
            return f"已标记完成:{self.items[index - 1]['text']}"

        elif action == "delete":
            if index is None or index < 1 or index > len(self.items):
                return "错误:无效的索引"
            removed = self.items.pop(index - 1)
            return f"已删除:{removed['text']}"

        return "未知操作"

现在我们有了一个可扩展的工具系统。接下来将这些工具集成到 Agent 中:

# 添加工具管理器

class ToolManager:
    """工具管理器:负责注册、调用和管理工具"""

    def __init__(self):
        """初始化工具管理器"""
        self.tools: Dict[str, BaseTool] = {}

    def register_tool(self, tool: BaseTool) -> None:
        """
        注册一个新工具

        Args:
            tool: 工具实例
        """
        self.tools[tool.name] = tool
        print(f"已注册工具: {tool.name}")

    def get_tool_schemas(self) -> List[Dict[str, Any]]:
        """
        获取所有工具的描述,供语言模型理解如何使用

        Returns:
            工具描述列表(OpenAI function calling 格式)
        """
        schemas = []
        for tool in self.tools.values():
            schema = {
                "type": "function",
                "function": {
                    "name": tool.name,
                    "description": tool.description,
                    "parameters": tool.parameters
                }
            }
            schemas.append(schema)
        return schemas

    def execute_tool(self, tool_name: str, arguments: Dict[str, Any]) -> str:
        """
        执行指定工具

        Args:
            tool_name: 工具名称
            arguments: 工具参数

        Returns:
            工具执行结果
        """
        if tool_name not in self.tools:
            return f"错误:未找到工具 '{tool_name}'"

        try:
            tool = self.tools[tool_name]
            result = tool.execute(**arguments)
            return result
        except Exception as e:
            return f"工具执行错误:{str(e)}"

class ToolEnabledAgent(AgentWithMemory):
    """支持工具调用的增强版 Agent"""

    def __init__(self, name: str, system_prompt: str, model: str = "gpt-4"):
        """初始化支持工具的 Agent"""
        super().__init__(name, system_prompt, model)
        self.tool_manager = ToolManager()
        self._register_default_tools()

    def _register_default_tools(self) -> None:
        """注册默认工具集"""
        self.tool_manager.register_tool(CalculatorTool())
        self.tool_manager.register_tool(DateTimeTool())
        self.tool_manager.register_tool(TodoTool())

    def think(self, user_input: str) -> str:
        """
        支持工具调用的思考方法

        Args:
            user_input: 用户输入

        Returns:
            Agent 的最终回复
        """
        self.add_message("user", user_input)

        max_iterations = 10  # 防止无限循环
        iteration = 0

        while iteration < max_iterations:
            iteration += 1

            # 获取当前可用的工具描述
            tools = self.tool_manager.get_tool_schemas()

            # 调用语言模型,让它决定是回复还是调用工具
            response = self._call_llm_with_tools(tools)

            if response.get("finish_reason") == "stop":
                # 模型决定结束对话
                final_response = response["message"]["content"]
                self.add_message("assistant", final_response)
                return final_response

            elif response.get("finish_reason") == "tool_calls":
                # 模型决定调用工具
                tool_calls = response["message"].get("tool_calls", [])

                # 处理所有工具调用
                for tool_call in tool_calls:
                    tool_name = tool_call["function"]["name"]
                    arguments = json.loads(tool_call["function"]["arguments"])

                    # 执行工具
                    tool_result = self.tool_manager.execute_tool(tool_name, arguments)

                    # 将工具调用和结果添加到对话历史
                    self.add_message("assistant", f"[调用工具: {tool_name}]")
                    self.add_message("tool", json.dumps({
                        "tool": tool_name,
                        "arguments": arguments,
                        "result": tool_result
                    }, ensure_ascii=False))

        return "处理超时,请尝试简化您的问题"

    def _call_llm_with_tools(self, tools: List[Dict]) -> Dict:
        """
        调用语言模型,并提供工具描述

        Args:
            tools: 工具描述列表

        Returns:
            模型响应
        """
        messages = self.get_conversation_history()

        # 模拟 API 调用(实际使用时替换为真实 API 调用)
        # 这里的实现省略了真实的 API 调用代码
        # 实际项目中需要使用 OpenAI 的 chat.completions.create 方法

        return {
            "finish_reason": "stop",
            "message": {"content": "工具调用功能需要配置真实的 API 密钥"}
        }

第五步:完整的示例程序

现在让我们把这些组件组合起来,创建一个完整可运行的示例:

"""
完整的 AI Agent 示例程序
演示了 Agent 的所有核心功能:对话、记忆、工具调用
"""

def main():
    """主函数:演示 Agent 的完整功能"""

    # 定义 Agent 的系统提示
    system_prompt = """你是一个智能助手,名为 Mia。
    你的特点是友好、耐心、知识渊博。
    你可以帮助用户回答问题、进行计算、管理待办事项等。
    当用户询问信息时,尽量提供准确、有用的答案。
    如果需要执行特定操作(如计算),请使用适当的工具。"""

    # 创建 Agent 实例
    agent = ToolEnabledAgent(
        name="Mia",
        system_prompt=system_prompt,
        model="gpt-4"
    )

    print("=" * 50)
    print("欢迎使用 AI Agent 演示程序")
    print("=" * 50)
    print("\nAgent 已准备就绪,可以开始对话!")
    print("输入 'quit' 或 'exit' 结束对话\n")

    # 简单的交互循环
    while True:
        try:
            user_input = input("你: ").strip()

            if user_input.lower() in ["quit", "exit", "退出"]:
                print("\n感谢使用,再见!")
                break

            if not user_input:
                continue

            # 获取 Agent 回复
            response = agent.think(user_input)
            print(f"\nMia: {response}\n")

            # 显示记忆状态(可选)
            if input("查看记忆状态? (y/n): ").lower() == "y":
                print(agent.memory.get_context_summary())
                print()

        except KeyboardInterrupt:
            print("\n\n对话被中断,感谢使用!")
            break
        except Exception as e:
            print(f"发生错误: {str(e)}")

if __name__ == "__main__":
    main()

常见应用场景与实战案例

场景一:个人助手应用

AI Agent 最直接的应用就是个人助手。你可以基于教程中的代码,构建一个能够帮你管理日程、回答问题、处理邮件的个人助理。

首先,扩展工具集以支持更多功能。你可以添加日历集成、邮件收发、天气查询等工具。每个工具都是一个独立的类,按照上面示例的模式实现即可。

其次,定制 Agent 的性格和行为。你可以通过修改系统提示词来调整 Agent 的回复风格。比如,如果你想要一个严肃专业的助手,可以在系统提示中强调”使用正式语言”;如果你想要一个轻松活泼的助手,可以添加”适当使用表情符号,保持对话轻松愉快”这样的要求。

最后,添加持续学习能力。通过完善长期记忆系统,让 Agent 能够记住你的偏好、习惯和重要信息。这样随着使用时间的增长,Agent 会变得越来越了解你,提供越来越个性化的服务。

场景二:自动化工作流

AI Agent 的另一个强大应用是自动化复杂的工作流程。传统上,如果我们要自动化一个包含多个步骤的任务,需要编写大量的脚本和逻辑判断。而有了 Agent,我们可以将任务描述给 Agent,让它自动规划并执行各个步骤。

例如,你可以构建一个自动代码审查助手。这个助手接收一段代码作为输入,然后自动进行以下步骤:首先理解代码的功能和结构;然后检查常见的代码问题,如潜在的 bug、安全漏洞、代码风格不一致等;最后生成详细的审查报告和改进建议。

实现这个功能需要添加一个代码分析工具,它能够调用静态分析器、语法检查器等外部工具。然后在 Agent 的系统提示中明确定义代码审查的标准和流程。Agent 会根据这些指引自主决定调用哪些工具、如何分析结果。

场景三:智能客服系统

将 AI Agent 用于客户服务是另一个热门应用场景。与传统的规则匹配式客服不同,基于 Agent 的客服能够理解用户的自然语言输入,处理复杂的、多轮对话的场景。

构建客服 Agent 时,关键是设计好知识库和回复策略。知识库中存储了产品信息、常见问题解答、政策说明等内容。Agent 根据用户的问题检索相关知识,生成准确的回复。

教程中详细介绍了如何实现知识检索功能。通过向量数据库,Agent 可以根据语义相似度找到最相关的知识条目,即使问题表述与知识库中的原文不完全一致也能准确匹配。

另外,客服 Agent 还需要处理一些特殊情况,如用户表达不满、询问敏感信息、遇到无法解决的问题等。教程中提供了处理这些情况的最佳实践,包括何时转人工、如何礼貌地拒绝回答、如何收集用户反馈等。


进阶技巧与最佳实践

提示词工程

编写高质量的系统提示词是发挥 Agent 能力的关键。以下是一些经过验证的最佳实践。

首先是角色定义要清晰明确。不要只说”你是一个助手”,而是详细描述助手的身份、专业背景、沟通风格和服务范围。比如,”你是一个有着十年经验的高级软件工程师,擅长 Python 和 JavaScript 开发。你习惯用简洁明了的语言解释技术概念,喜欢通过代码示例来帮助他人理解。”

其次是输出格式要具体规范。如果需要 Agent 返回结构化的数据,应该在提示词中明确说明格式要求。例如,”请按以下格式返回结果:\n1. 问题描述:…\n2. 可能原因:…\n3. 解决方案:…\n4. 预防措施:…”

第三是要包含边界条件。明确告诉 Agent 什么能做、什么不能做、遇到边界情况应该如何处理。这可以避免 Agent 产生不恰当的回复或执行危险的操作。

错误处理与容错

健壮的错误处理是一个可靠 Agent 系统的必备要素。教程中演示了多层错误处理策略。

在工具层面,每个工具都应该捕获并处理可能的异常。比如,计算器工具需要处理除零错误、无效表达式错误;网络请求工具需要处理超时、连接失败等异常。处理错误时,应该返回有意义的错误信息,而不是让程序崩溃。

在 Agent 层面,应该实现重试机制和降级策略。当一个工具调用失败时,Agent 可以尝试使用替代方案,或者简化请求后重试。如果所有方案都失败,应该优雅地向用户解释情况,而不是表现出手足无措。

在系统层面,要有监控和日志记录机制。记录 Agent 的所有操作和决策,便于问题排查和性能优化。对于生产环境,还需要设置报警机制,在出现异常时及时通知运维人员。

性能优化

当 Agent 变得越来越复杂时,性能优化就变得很重要。以下是一些实用的优化技巧。

对话历史的压缩和截断是最常见的优化手段。随着对话的进行,历史消息会越来越长,这会导致 API 调用成本增加和响应延迟。可以实现智能截断策略,比如只保留最近 N 轮对话,或者删除中间不重要的话题轮次。

工具调用的批处理也能显著提升性能。如果 Agent 需要执行多个独立的任务,尽量将它们合并为一次批量调用,而不是依次执行多次。例如,如果 Agent 需要查询多个数据源,可以设计一个批量查询工具,同时获取所有数据。

异步处理和流式输出能够改善用户体验。对于耗时的操作,使用异步方式处理,让 Agent 能够边思考边向用户展示进度。流式输出则让用户能够实时看到 Agent 的回复,而不必等待完整的响应生成完毕。

安全考虑

AI Agent 系统中存在一些特有的安全问题,需要认真对待。

首先是提示词注入攻击。恶意用户可能试图通过精心构造的输入来劫持 Agent 的行为,使其执行未授权的操作。防范措施包括:对用户输入进行清理和验证;在系统提示中明确边界;限制工具的权限范围。

其次是敏感信息泄露。Agent 可能会在回复中不小心暴露它不应该知道的信息。处理方法是实施严格的信息隔离策略,确保 Agent 只能访问它应该访问的数据。

第三是工具滥用。即使是善意的工具,如果使用不当也可能造成问题。比如,一个文件操作工具可能被用来访问或删除重要文件。应该在工具层面实施权限控制和操作审计。


实际项目中的集成方案

与 Web 应用集成

将 AI Agent 集成到 Web 应用中是最常见的需求之一。教程中提供了使用 FastAPI 构建 API 服务的完整示例。

首先,需要将 Agent 封装为一个异步服务。这样可以让应用同时处理多个用户请求,而不会因为 Agent 的处理时间而阻塞。下面的代码展示了基本的服务封装:

"""
AI Agent API 服务示例
使用 FastAPI 构建 RESTful API
"""

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional, List
import uvicorn

# 定义请求和响应模型
class ChatRequest(BaseModel):
    """聊天请求模型"""
    session_id: str  # 会话 ID,用于保持对话上下文
    message: str     # 用户消息
    user_id: Optional[str] = None  # 用户 ID(可选)

class ChatResponse(BaseModel):
    """聊天响应模型"""
    session_id: str
    message: str
    agent_name: str
    tools_used: List[str] = []

# 创建 FastAPI 应用
app = FastAPI(
    title="AI Agent API",
    description="提供 AI Agent 对话服务的 API",
    version="1.0.0"
)

# 会话管理器:存储不同用户的对话状态
class SessionManager:
    def __init__(self):
        self.sessions = {}

    def get_or_create_agent(self, session_id: str) -> ToolEnabledAgent:
        """获取或创建指定会话的 Agent 实例"""
        if session_id not in self.sessions:
            # 创建新的 Agent 实例
            system_prompt = """你是一个乐于助人的 AI 助手。"""
            agent = ToolEnabledAgent(
                name="Assistant",
                system_prompt=system_prompt
            )
            self.sessions[session_id] = agent
        return self.sessions[session_id]

    def remove_session(self, session_id: str) -> None:
        """删除指定的会话"""
        if session_id in self.sessions:
            del self.sessions[session_id]

# 全局会话管理器
session_manager = SessionManager()

@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
    """
    处理用户聊天请求

    Args:
        request: 聊天请求,包含会话 ID 和消息

    Returns:
        Agent 的回复
    """
    try:
        # 获取或创建 Agent
        agent = session_manager.get_or_create_agent(request.session_id)

        # 处理用户消息
        response = agent.think(request.message)

        # 返回响应
        return ChatResponse(
            session_id=request.session_id,
            message=response,
            agent_name=agent.name
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.delete("/session/{session_id}")
async def end_session(session_id: str):
    """结束指定的会话"""
    session_manager.remove_session(session_id)
    return {"message": "会话已结束"}

@app.get("/health")
async def health_check():
    """健康检查端点"""
    return {"status": "healthy", "active_sessions": len(session_manager.sessions)}

# 启动服务
if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

运行这个服务后,你可以通过 HTTP 请求与 Agent 交互,非常适合与前端应用或移动应用集成。

与数据库集成

持久化存储是生产环境中的必要功能。教程演示了如何将 Agent 与不同的数据存储方案集成。

对于对话历史的存储,可以使用 PostgreSQL 或 MySQL 等关系型数据库。设计表结构时,需要记录会话 ID、消息内容、消息角色、时间戳等信息。为了提高检索效率,可以添加消息索引。

对于向量存储,如果 Agent 需要快速检索大量文档中的相关内容,可以使用专门的向量数据库如 Milvus、Qdrant 或 Pinecone。将文档转换为向量后存储,需要时根据语义相似度进行检索。

对于缓存层,可以使用 Redis 来加速频繁访问的数据,如用户配置、最近的对话历史等。合理的缓存策略能够显著减少数据库负载和响应时间。

与消息队列集成

在需要处理大量并发请求的场景中,消息队列是必不可少的组件。教程中介绍了如何将 Agent 与 RabbitMQ 或 Kafka 集成。

消息队列的基本工作流程是:用户的请求首先被发送到队列中;后台的 Worker 进程从队列中取出请求,调用 Agent 进行处理;处理完成后,将结果发送到另一个队列或直接返回给客户端。

这种架构的优势在于:可以削峰填谷,平滑处理突发的请求高峰;可以实现服务的水平扩展,通过增加 Worker 数量来提高吞吐量;可以提高系统的可靠性,即使某个组件暂时不可用,请求也不会丢失。


总结与延伸学习

通过这篇教程,你应该已经对 AI Agent 有了全面的理解,并掌握了开发一个基本 Agent 的核心技能。我们从最基础的 Agent 结构开始,逐步添加了语言模型集成、记忆系统、工具调用等关键功能,最终实现了一个功能完整的智能助手。

学习是无止境的。微软的这个教程只是入门,想要成为 AI Agent 领域的专家,还有很多内容等待探索。以下是一些延伸学习方向:

如果你对多模态 Agent 感兴趣,可以学习如何让 Agent 处理图像、音频、视频等多种类型的数据。这涉及到多模态大模型的使用,如 GPT-4V、Claude 的视觉能力等。

如果你关注 Agent 的自主决策能力,可以深入研究强化学习和规划算法。了解如何让 Agent 在复杂环境中自主探索和学习,而不是仅仅依赖预定义的规则。

如果你想构建更强大的 Agent 系统,可以学习 Agent 协作框架,如 AutoGen、CrewAI 等。这些框架提供了多 Agent 协作、任务分解与分配等高级功能。

最后,不要忘记关注这个领域的最新发展。AI Agent 技术日新月异,新的模型、新的框架、新的应用场景不断涌现。保持学习的心态,多阅读论文、多参与社区讨论、多动手实践,你一定能在这个激动人心的领域中找到自己的位置。


相关资源链接

官方资源

微软 AI Agents for Beginners 官方仓库:https://github.com/microsoft/ai-agents-for-beginners

微软官方文档中心:https://learn.microsoft.com/

Azure OpenAI Service 文档:https://learn.microsoft.com/azure/ai-services/openai/

延伸学习项目

LangChain:构建 LLM 应用的流行框架 https://github.com/langchain-ai/langchain

AutoGen:微软开源的多 Agent 协作框架 https://github.com/microsoft/autogen

CrewAI:多 Agent 协作编排框架 https://github.com/joaomdmoura/crewAI

LlamaIndex:知识检索增强生成框架 https://github.com/run-llama/llama_index

工具与服务

OpenAI API:https://platform.openai.com/

Hugging Face:机器学习模型和数据集平台 https://huggingface.co/

Qdrant:开源向量数据库 https://github.com/qdrant/qdrant

Redis:高性能缓存和消息队列 https://redis.io/

如果内容对您有帮助,欢迎打赏

您的支持是我继续创作的动力

前往打赏页面

评论区

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注

**微软开源AI Agent实战项目,零基础也能快速构建智能代理(附完整教程)**

**微软开源AI Agent实战项目,零基础也能快速构建智能代理(附完整教程)**

微软开源AI Agent实战项目,零基础也能快速构建智能代理(附完整教程)

为什么你还在为AI Agent开发发愁?微软官方出品的这套入门项目,可能是目前最值得学习的资源


为什么值得关注 / 微软这个项目到底解决什么问题

在人工智能快速发展的今天,AI Agent(人工智能代理)已经成为技术领域最热门的话题之一。简单来说,AI Agent是一种能够自主理解目标、规划行动、执行任务并根据反馈进行自我调整的智能系统。它不同于传统的AI助手只能被动回答问题,而是能够主动规划路径、调用工具、与其他代理协作,最终完成复杂的多步骤任务。

然而,对于想要学习AI Agent开发的开发者来说,入门门槛并不低。市场上的资料分散、示例代码质量参差不齐、缺乏系统性的学习路径,这些问题让很多初学者望而却步。正是为了解决这些痛点,微软推出了这个名为”AI Agents for Beginners”的官方教程项目。

微软AI Agents for Beginners项目具有以下几个显著特点,使其成为学习AI Agent开发的理想选择:

第一,权威性与系统性兼备。 作为科技巨头微软的官方项目,这个教程不仅代表了业界最佳实践,还提供了完整的学习路径,从基础概念到高级应用,循序渐进,非常适合零基础学习者。

第二,多语言支持。 项目提供了Python、JavaScript/TypeScript和C#三种主流编程语言的示例代码,无论你熟悉哪种语言,都能找到适合自己的学习路径。这种多语言支持在同类项目中是相当罕见的。

第三,涵盖主流框架。 教程中包含了Semantic Kernel、AutoGen等业界领先的AI Agent开发框架,让学习者能够接触到真实的工业级工具,而不是纸上谈兵的概念讲解。

第四,理论与实践深度融合。 每个知识点都配有可运行的代码示例,学习者可以立即动手实践,通过实际运行代码来加深理解,而不是被动地阅读理论。

第五,持续更新维护。 作为微软的官方项目,它会随着技术发展持续更新,确保学习内容与时俱进,不会出现知识过时的问题。


环境搭建 / 准备工作

在开始学习之前,我们需要先搭建好开发环境。微软AI Agents for Beginners项目对环境的要求并不苛刻,但为了确保顺利运行所有示例代码,我们需要做好以下准备工作。

硬件和操作系统要求

项目可以在Windows、macOS和Linux三大主流操作系统上运行。硬件方面,建议至少具备8GB内存,如果计划运行较大的语言模型,可能需要16GB或更高。磁盘空间方面,代码仓库本身只需要几百MB,但运行各种示例可能需要额外的存储空间来存放依赖包和模型文件。

安装Python环境

对于Python学习者,需要准备Python 3.10或更高版本。可以通过以下步骤检查和安装Python:

打开终端或命令提示符,输入以下命令检查Python版本:

python3 --version

如果显示的版本低于3.10,或者提示命令未找到,需要前往Python官网下载并安装最新版本的Python。安装过程中记得勾选”Add Python to PATH”选项,这样可以在任何目录下直接调用Python命令。

安装完成后,可以使用pip来管理Python包。建议先升级pip到最新版本:

python3 -m pip install --upgrade pip

安装Node.js环境

对于JavaScript/TypeScript学习者,需要Node.js 18或更高版本。可以通过以下命令检查是否已安装:

node --version
npm --version

如果需要安装,推荐使用nvm(Node Version Manager)来管理Node.js版本,这样可以方便地在不同版本之间切换。

安装.NET SDK(C#学习者)

如果计划学习C#示例,需要安装.NET 8.0或更高版本。可以通过以下命令检查:

dotnet --version

可以从.NET官网下载安装包进行安装。

获取API密钥

大多数AI Agent应用需要调用大语言模型的API。常见的选项包括OpenAI的GPT系列模型、Azure OpenAI服务等。项目本身设计为支持多种模型接入,但建议你准备至少一个API密钥来运行示例代码。

对于OpenAI API,需要前往OpenAI官网注册账号并获取API密钥。需要注意API使用是按量计费的,新用户通常会获得一定的免费额度。

对于Azure OpenAI服务,需要在Azure门户中创建资源并获取相应的密钥和端点。

克隆项目仓库

所有准备工作完成后,就可以获取微软的官方教程代码了。打开终端,执行以下命令克隆仓库:

git clone https://github.com/microsoft/ai-agents-for-beginners.git

克隆完成后,进入项目目录:

cd ai-agents-for-beginners

建议先浏览一下项目结构,了解各个文件夹和文件的作用。项目通常会按照不同的主题或框架组织代码,每个主题下都有详细的README文档说明。

创建虚拟环境(推荐)

为了避免依赖冲突,建议为Python项目创建独立的虚拟环境。可以使用venv模块:

python3 -m venv agent-env

激活虚拟环境的方法因操作系统而异。在Windows上:

agent-env\Scripts\activate

在macOS和Linux上:

source agent-env/bin/activate

激活后,终端提示符会显示虚拟环境名称,表示你已经进入了隔离的开发环境。在这个环境中安装的包不会影响系统全局的Python环境。


核心功能详解 / 从概念到实现

微软AI Agents for Beginners项目涵盖了AI Agent开发的多个核心方面。深入理解这些核心概念和功能,是掌握AI Agent开发的基础。

什么是AI Agent

在深入代码之前,我们需要先从概念层面理解什么是AI Agent。一个完整的AI Agent通常由以下几个核心组件构成:

感知模块(Perception)负责接收和解析来自外部的输入,可以是文本、图像、音频等多种模态。在代码层面,这通常体现为对用户输入或系统事件的处理逻辑。

规划模块(Planning)是AI Agent的”大脑”,负责将复杂任务分解为可执行的子步骤,并确定执行顺序。这涉及到推理能力,Agent需要理解任务目标,分析当前状态,规划出一条可行的行动路径。

记忆模块(Memory)让Agent能够存储和检索信息,包括短期记忆(当前会话中的上下文)和长期记忆(跨会话积累的知识)。记忆模块是实现连贯对话和持续学习的基础。

工具使用(Tool Use)能力使Agent能够调用外部系统、执行特定操作,如查询数据库、调用API、读写文件等。这是Agent区别于纯语言模型的关键特性。

行动执行(Action)模块负责实际执行规划模块确定的行动,可能涉及调用API、生成响应、操作环境等。

Semantic Kernel框架详解

Semantic Kernel是微软推出的轻量级AI应用编程框架,它巧妙地将大语言模型与传统软件开发结合在一起。框架的核心设计理念是”技能(Skills)”和”连接器(Connectors)”。

技能(Skills)是Semantic Kernel中的核心概念,代表Agent可以执行的能力。一个技能由三个部分组成:描述(manifest)、触发条件(triggers)和实现逻辑。描述定义了技能的名称、功能和参数;触发条件决定了何时调用该技能;实现逻辑则是具体的代码执行逻辑。

例如,一个”天气查询”技能可能包含以下结构:

天气查询技能
├── 名称: WeatherQuery
├── 功能: 根据城市名称查询天气信息
├── 参数: 城市名(必填)
└── 返回: 天气信息字符串

连接器(Connectors)则提供了与外部服务交互的能力。Semantic Kernel内置了多种连接器,可以方便地接入OpenAI、Azure OpenAI、Hugging Face等大语言模型服务。

在Semantic Kernel中,编排(Orchestration)是实现复杂任务的关键机制。通过编排器,Agent可以按照预定义的流程或动态生成的计划,依次调用不同的技能来完成任务。这种设计既保证了系统的可控性,又保留了一定的灵活性。

AutoGen框架详解

AutoGen是微软另一个重要的AI Agent框架,它的核心特点是支持多代理协作和对话式编程。与Semantic Kernel相比,AutoGen更加强调代理之间的交互和协作。

AutoGen中的代理(Agent)是一个独立的实体,拥有自己的系统提示(system prompt)、角色定义和行为逻辑。每个代理都可以主动发起对话、响应其他代理的消息、调用工具或执行代码。

会话管理(Conversation)是AutoGen的另一个核心概念。框架自动管理代理之间的消息传递和上下文维护,开发者只需要定义代理的行为和交互规则,剩下的通信细节由框架处理。

AutoGen的一个显著优势是支持人机协作。在实际应用场景中,有时需要人类在关键时刻介入决策。AutoGen提供了自然的方式来实现这种人机协作,代理可以在需要时暂停执行,等待人类输入后再继续。

多代理模式是AutoGen的亮点之一。通过定义不同角色的代理(如用户代理、助手代理、评审代理等),可以让它们分工协作完成复杂任务。这种模式模拟了真实世界中的团队协作,每个代理专注于自己的职责,通过对话协商达成最终目标。

工具调用与函数执行

让AI Agent真正具有实用价值的关键,是它能够调用外部工具和执行具体操作。微软的教程详细介绍了多种工具调用的实现方式。

代码执行工具允许Agent执行Python或其他语言的代码。这对于数据分析、数学计算、文本处理等任务特别有用。Agent可以生成代码、提交执行、获取结果、基于结果进行调整。

搜索工具使Agent能够查询外部信息源,包括网页搜索、知识库查询、数据库检索等。这扩展了Agent的知识范围,使其能够获取实时信息。

文件操作工具让Agent可以读写本地文件,进行文档处理、数据导入导出等操作。

API调用工具允许Agent与各种Web服务交互,获取天气、新闻、股票行情等外部数据。

在实现层面,工具调用通常遵循”描述-调用-响应”的模式。首先,需要为每个工具定义清晰的描述,包括工具名称、功能说明、参数规范和返回值格式。然后,语言模型根据用户请求和可用工具描述,决定是否调用工具以及调用哪些工具。最后,工具执行的结果被返回给模型,模型据此生成最终响应。


实战教程 / 一步步构建你的第一个AI Agent

理论讲解已经足够多了,现在让我们通过具体的代码示例,一步步构建一个功能完整的AI Agent。我们将以Python和Semantic Kernel为主要工具,带领读者从零开始实现一个能够回答问题、执行代码、搜索信息的智能代理。

第一步:创建基础项目结构

首先,创建一个新的Python项目。在克隆的仓库中,创建一个新的文件夹用于我们的实战练习:

mkdir my-first-agent
cd my-first-agent

在文件夹中创建以下文件结构:

my-first-agent/
├── config.json          # 配置文件
├── requirements.txt     # 依赖列表
└── agent.py            # 主程序文件

config.json用于存储API密钥和配置信息:

{
    "api_key": "your-openai-api-key-here",
    "model": "gpt-4",
    "base_url": "https://api.openai.com/v1"
}

requirements.txt列出项目依赖:

semantic-kernel>=0.9.0
openai>=1.0.0
python-dotenv>=1.0.0
requests>=2.31.0

然后安装依赖:

pip install -r requirements.txt

第二步:初始化Semantic Kernel

现在创建agent.py文件,开始编写我们的AI Agent。首先是基础的初始化代码:

"""
第一个AI Agent - 基础版本
本示例演示如何初始化Semantic Kernel并创建简单的聊天代理
"""

import os
from dotenv import load_dotenv
import semantic_kernel as sk
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion

# 加载环境变量
load_dotenv()

# 获取API密钥
api_key = os.getenv("OPENAI_API_KEY") or os.getenv("API_KEY")
if not api_key:
    raise ValueError("请设置API密钥到环境变量或.env文件中")

# 创建语义内核实例
kernel = sk.Kernel()

# 配置聊天服务
kernel.add_chat_service(
    "chat-gpt",
    OpenAIChatCompletion(
        ai_model_id="gpt-4",
        api_key=api_key,
    )
)

print("=== Semantic Kernel 初始化完成 ===")
print(f"已配置模型: gpt-4")

这段代码完成了最基础的初始化工作:加载配置、创建内核实例、配置聊天服务。运行这段代码,如果看到初始化完成的消息,说明基础配置正确。

第三步:定义技能(Skills)

现在我们为Agent添加实际的能力。在Semantic Kernel中,技能是通过特定格式的目录结构定义的。让我们创建一个简单的技能:

"""
定义AI Agent的技能
技能是Agent执行特定任务的能力单元
"""

from semantic_kernel.core_skills import MathSkill, TextSkill, FileIOSkill

# 添加内置技能到内核
kernel.import_skill(MathSkill(), skill_name="math")
kernel.import_skill(TextSkill(), skill_name="text")
kernel.import_skill(FileIOSkill(), skill_name="file_io")

print("=== 已加载内置技能 ===")
print("可用技能: math (数学计算), text (文本处理), file_io (文件操作)")

除了使用内置技能,我们还可以定义自定义技能。创建一个新的技能目录结构:

skills/
└── AssistantSkill/
    └── skprompt.txt

skprompt.txt定义技能的提示词和参数:

你是一个乐于助人的AI助手。请根据用户的问题给出清晰、准确的回答。

用户的问题是: {{$input}}

请用简洁易懂的语言回答。如果不确定答案,请诚实地说明。

在代码中加载自定义技能:

from semantic_kernel.skill_library import SemanticSkill

# 加载自定义技能
assistant_skill = kernel.import_skill(
    SemanticSkill.from_directory("skills/AssistantSkill", kernel),
    skill_name="assistant"
)

print("=== 自定义技能加载成功 ===")

第四步:创建对话循环

现在我们需要一个主循环,让Agent能够与用户持续对话:

"""
主对话循环
实现用户与AI Agent的持续交互
"""

def run_conversation():
    """运行对话循环"""

    print("=" * 60)
    print("欢迎使用 AI Agent 助手!")
    print("输入你的问题,或者输入 '退出' 结束对话")
    print("=" * 60)
    print()

    # 创建聊天历史
    chat_history = []

    while True:
        # 获取用户输入
        user_input = input("你: ").strip()

        # 检查退出条件
        if user_input.lower() in ["退出", "exit", "quit"]:
            print("\n感谢使用,再见!")
            break

        # 空输入处理
        if not user_input:
            print("请输入内容后再试。")
            continue

        # 将用户消息添加到历史
        chat_history.append({"role": "user", "content": user_input})

        try:
            # 调用AI生成回复
            response = kernel.invoke(
                "assistant",
                input=user_input
            )

            # 显示AI回复
            ai_response = str(response)
            print(f"\nAI: {ai_response}\n")

            # 将AI回复添加到历史
            chat_history.append({"role": "assistant", "content": ai_response})

        except Exception as e:
            print(f"\n抱歉,处理你的问题时遇到了错误: {e}\n")

# 运行对话
if __name__ == "__main__":
    run_conversation()

第五步:添加代码执行能力

一个强大的AI Agent应该能够执行代码。让我们扩展我们的Agent,添加代码执行能力:

"""
代码执行技能
让Agent能够运行Python代码并返回结果
"""

import subprocess
import sys
from io import StringIO
from contextlib import redirect_stdout, redirect_stderr

def execute_python_code(code: str) -> str:
    """
    安全地执行Python代码

    参数:
        code: 要执行的Python代码字符串

    返回:
        执行结果的字符串描述
    """

    # 代码执行前的准备
    output = StringIO()
    error_output = StringIO()

    try:
        # 使用上下文管理器捕获输出
        with redirect_stdout(output), redirect_stderr(error_output):
            # 创建隔离的命名空间执行代码
            exec_globals = {"__name__": "__main__"}
            exec(code, exec_globals)

        # 获取捕获的输出
        result = output.getvalue()
        errors = error_output.getvalue()

        # 构建返回结果
        if result:
            return f"执行成功!\n输出结果:\n{result}"
        elif errors:
            return f"代码执行完成,但有一些警告:\n{errors}"
        else:
            return "代码执行完成,没有输出"

    except SyntaxError as e:
        return f"语法错误: {e}"
    except Exception as e:
        return f"执行错误: {type(e).__name__}: {e}"

# 测试代码执行功能
if __name__ == "__main__":
    test_code = """
# 测试代码执行功能
numbers = [1, 2, 3, 4, 5]
squared = [x ** 2 for x in numbers]
print(f"原始列表: {numbers}")
print(f"平方后: {squared}")
print(f"总和: {sum(squared)}")
"""

    print("=== 测试代码执行功能 ===")
    result = execute_python_code(test_code)
    print(result)

第六步:实现工具调用功能

现在让我们实现一个更高级的功能——让AI Agent能够根据需要自主调用工具:

"""
工具注册与自动调用系统
让AI Agent能够智能选择和使用工具
"""

import json
import re
from typing import Dict, List, Callable, Any

class ToolManager:
    """工具管理器 - 负责注册和调度工具"""

    def __init__(self):
        # 存储可用工具
        self.tools: Dict[str, Callable] = {}
        # 存储工具描述
        self.tool_descriptions: Dict[str, str] = {}

    def register_tool(self, name: str, func: Callable, description: str):
        """
        注册一个新工具

        参数:
            name: 工具名称
            func: 工具函数
            description: 工具的功能描述
        """
        self.tools[name] = func
        self.tool_descriptions[name] = description
        print(f"工具已注册: {name}")

    def get_available_tools(self) -> str:
        """获取所有可用工具的描述"""
        if not self.tools:
            return "当前没有可用工具"

        description = "可用工具:\n"
        for name, desc in self.tool_descriptions.items():
            description += f"- {name}: {desc}\n"
        return description

    def call_tool(self, tool_name: str, **kwargs) -> str:
        """
        调用指定工具

        参数:
            tool_name: 工具名称
            **kwargs: 传递给工具的参数

        返回:
            工具执行结果
        """
        if tool_name not in self.tools:
            return f"错误: 工具 '{tool_name}' 不存在"

        try:
            result = self.tools[tool_name](**kwargs)
            return str(result)
        except Exception as e:
            return f"工具执行错误: {e}"

# 创建工具管理器实例
tool_manager = ToolManager()

# 注册各种工具
def search_wikipedia(query: str) -> str:
    """模拟维基百科搜索"""
    # 实际项目中可以调用真实的Wikipedia API
    return f"关于 '{query}' 的信息: 这是一个模拟的搜索结果。在实际应用中,这里会返回真实的搜索结果。"

def calculate(expression: str) -> str:
    """数学计算工具"""
    try:
        # 安全评估简单数学表达式
        # 注意:实际应用中应该使用更安全的评估方式
        allowed_chars = set("0123456789+-*/(). ")
        if all(c in allowed_chars for c in expression):
            result = eval(expression)
            return f"计算结果: {expression} = {result}"
        else:
            return "错误: 表达式包含无效字符"
    except Exception as e:
        return f"计算错误: {e}"

def get_current_time() -> str:
    """获取当前时间"""
    from datetime import datetime
    now = datetime.now()
    return f"当前时间: {now.strftime('%Y年%m月%d日 %H:%M:%S')}"

# 注册工具
tool_manager.register_tool(
    "search",
    search_wikipedia,
    "搜索维基百科,输入搜索关键词"
)

tool_manager.register_tool(
    "calculate",
    calculate,
    "进行数学计算,输入数学表达式"
)

tool_manager.register_tool(
    "time",
    get_current_time,
    "获取当前日期和时间"
)

print("\n=== 工具系统已初始化 ===")
print(tool_manager.get_available_tools())

第七步:构建完整的Agent类

现在让我们把所有组件整合到一个完整的Agent类中:

"""
完整的AI Agent实现
整合所有功能到一个连贯的系统中
"""

import os
import json
from typing import Dict, List, Optional
from dataclasses import dataclass, field
from datetime import datetime

@dataclass
class Message:
    """对话消息的数据类"""
    role: str  # "user", "assistant", "system"
    content: str
    timestamp: datetime = field(default_factory=datetime.now)
    tool_calls: List[Dict] = field(default_factory=list)

class AIAgent:
    """AI Agent主类"""

    def __init__(self, name: str = "AI助手", system_prompt: str = ""):
        """
        初始化AI Agent

        参数:
            name: Agent的名称
            system_prompt: 系统提示词
        """
        self.name = name
        self.system_prompt = system_prompt or self._default_system_prompt()
        self.conversation_history: List[Message] = []
        self.tool_manager = ToolManager()
        self._setup_tools()

        print(f"=== Agent '{name}' 已初始化 ===")

    def _default_system_prompt(self) -> str:
        """生成默认的系统提示词"""
        return f"""你是一个名为{self.name}的智能AI助手。

你有以下能力:
1. 回答各种问题
2. 执行Python代码
3. 搜索信息
4. 进行数学计算
5. 获取当前时间

当用户询问信息或执行任务时,你可以选择使用合适的工具来更好地帮助他们。

工具列表:
{self.tool_manager.get_available_tools()}
"""

    def _setup_tools(self):
        """设置Agent可用的工具"""
        self.tool_manager.register_tool(
            "code_executor",
            execute_python_code,
            "执行Python代码,输入代码字符串"
        )
        self.tool_manager.register_tool(
            "search",
            search_wikipedia,
            "搜索信息,输入搜索关键词"
        )
        self.tool_manager.register_tool(
            "calculate",
            calculate,
            "进行数学计算,输入数学表达式"
        )
        self.tool_manager.register_tool(
            "time",
            get_current_time,
            "获取当前日期和时间"
        )

    def process_request(self, user_input: str) -> str:
        """
        处理用户请求的主方法

        参数:
            user_input: 用户的输入

        返回:
            Agent的响应
        """
        # 添加用户消息到历史
        self.conversation_history.append(
            Message(role="user", content=user_input)
        )

        # 在实际项目中,这里会调用大语言模型
        # 这里我们用简化的逻辑来演示
        response = self._generate_response(user_input)

        # 添加助手回复到历史
        self.conversation_history.append(
            Message(role="assistant", content=response)
        )

        return response

    def _generate_response(self, user_input: str) -> str:
        """
        生成响应(简化版本)

        在实际项目中,这里会调用大语言模型API
        模型会根据上下文和可用工具生成智能响应
        """
        user_lower = user_input.lower()

        # 关键词匹配来决定使用哪个工具
        if any(word in user_lower for word in ["搜索", "查找", "找"]):
            # 提取搜索关键词
            keyword = user_input.split()[-1]
            return self.tool_manager.call_tool("search", query=keyword)

        elif any(word in user_lower for word in ["计算", "等于", "+", "-", "*", "/"]):
            # 尝试提取数学表达式
            import re
            expr = re.search(r'[\d\+\-\*/\(\)\.]+', user_input)
            if expr:
                return self.tool_manager.call_tool("calculate", expression=expr.group())
            return "请提供要计算的数学表达式"

        elif any(word in user_lower for word in ["时间", "几点", "日期"]):
            return self.tool_manager.call_tool("time")

        elif any(word in user_lower for word in ["代码", "运行", "执行"]):
            # 尝试提取代码块
            if "```python" in user_input:
                code = user_input.split("```python")[1].split("```")[0]
            elif "```" in user_input:
                code = user_input.split("```")[1].split("```")[0]
            else:
                code = user_input.replace("运行", "").replace("代码", "").strip()
            return self.tool_manager.call_tool("code_executor", code=code)

        else:
            # 默认回复
            return f"我收到了你的问题:'{user_input}'。作为一个AI助手,我可以帮你回答问题、执行代码、搜索信息或进行计算。请问有什么我可以帮助你的?"

    def get_conversation_summary(self) -> str:
        """获取对话摘要"""
        summary = f"\n{'='*60}\n"
        summary += f"与 {self.name} 的对话摘要\n"
        summary += f"{'='*60}\n"
        summary += f"对话轮数: {len(self.conversation_history) // 2}\n"
        summary += f"总消息数: {len(self.conversation_history)}\n"
        summary += "\n最近5条消息:\n"

        for msg in self.conversation_history[-5:]:
            role_display = "你" if msg.role == "user" else self.name
            summary += f"[{msg.timestamp.strftime('%H:%M:%S')}] {role_display}: {msg.content[:50]}...\n"

        return summary

# 创建并测试Agent
if __name__ == "__main__":
    # 创建Agent实例
    agent = AIAgent(name="小助手")

    # 测试各种功能
    print("\n=== 开始功能测试 ===\n")

    # 测试搜索
    print("测试1 - 信息搜索:")
    print(agent.process_request("搜索人工智能"))
    print()

    # 测试计算
    print("测试2 - 数学计算:")
    print(agent.process_request("计算 25 * 4 + 100"))
    print()

    # 测试时间
    print("测试3 - 获取时间:")
    print(agent.process_request("现在几点了"))
    print()

    # 测试代码执行
    print("测试4 - 代码执行:")
    test_code = """
import math
print(f"圆周率约等于: {math.pi}")
print(f"2的10次方等于: {2**10}")
"""
    print(agent.process_request("运行代码"))
    print()

    # 显示对话摘要
    print(agent.get_conversation_summary())

第八步:实现多代理协作

现在让我们实现更高级的功能——多个Agent协作完成任务。这是AutoGen框架的核心特性之一。

"""
多代理协作系统
演示多个AI Agent如何分工合作完成复杂任务
"""

from typing import List, Dict, Any
from dataclasses import dataclass
from enum import Enum

class AgentRole(Enum):
    """代理角色枚举"""
    PLANNER = "planner"      # 规划者 - 分析任务并制定计划
    EXECUTOR = "executor"   # 执行者 - 执行具体任务
    REVIEWER = "reviewer"   # 评审者 - 检查结果并提出建议
    COORDINATOR = "coordinator"  # 协调者 - 协调各代理工作

@dataclass
class Task:
    """任务数据类"""
    id: str
    description: str
    status: str = "pending"  # pending, in_progress, completed, failed
    assigned_to: str = ""
    result: Any = None

class MultiAgentSystem:
    """多代理协作系统"""

    def __init__(self):
        # 存储所有代理
        self.agents: Dict[str, AIAgent] = {}
        # 任务队列
        self.tasks: List[Task] = []
        # 任务计数器
        self.task_counter = 0

        print("=== 多代理系统已初始化 ===")

    def register_agent(self, agent_id: str, role: AgentRole, name: str):
        """
        注册一个新代理

        参数:
            agent_id: 代理的唯一标识
            role: 代理的角色
            name: 代理的名称
        """
        role_prompts = {
            AgentRole.PLANNER: "你是一个任务规划专家,擅长分析问题并制定详细的执行计划。",
            AgentRole.EXECUTOR: "你是一个任务执行专家,擅长高效准确地完成任务。",
            AgentRole.REVIEWER: "你是一个质量评审专家,擅长检查工作质量并提出改进建议。",
            AgentRole.COORDINATOR: "你是一个项目协调专家,擅长协调多个团队成员完成复杂项目。"
        }

        agent = AIAgent(
            name=name,
            system_prompt=role_prompts.get(role, "")
        )
        agent.role = role

        self.agents[agent_id] = agent
        print(f"已注册代理: {name} (角色: {role.value})")

    def create_task(self, description: str) -> str:
        """
        创建新任务

        参数:
            description: 任务描述

        返回:
            任务ID
        """
        self.task_counter += 1
        task_id = f"task_{self.task_counter}"

        task = Task(
            id=task_id,
            description=description
        )
        self.tasks.append(task)

        print(f"已创建任务: {task_id} - {description}")
        return task_id

    def assign_task(self, task_id: str, agent_id: str):
        """为任务分配执行代理"""
        for task in self.tasks:
            if task.id == task_id:
                task.assigned_to = agent_id
                task.status = "in_progress"
                print(f"任务 {task_id} 已分配给 {agent_id}")
                return True
        return False

    def execute_workflow(self, task_id: str) -> Dict[str, Any]:
        """
        执行完整的工作流程

        流程包括:
        1. 规划阶段 - 分析任务
        2. 执行阶段 - 完成任务
        3. 评审阶段 - 检查结果
        """

        # 找到任务
        task = None
        for t in self.tasks:
            if t.id == task_id:
                task = t
                break

        if not task:
            return {"error": f"任务 {task_id} 不存在"}

        results = {
            "task_id": task_id,
            "description": task.description,
            "steps": []
        }

        # 步骤1: 规划
        print("\n" + "="*60)
        print("阶段1: 任务规划")
        print("="*60)

        planner_response = self.agents.get("planner").process_request(
            f"请分析以下任务并制定执行计划:{task.description}"
        )
        results["steps"].append({
            "stage": "planning",
            "agent": "planner",
            "output": planner_response
        })
        print(f"规划结果: {planner_response[:100]}...")

        # 步骤2: 执行
        print("\n" + "="*60)
        print("阶段2: 任务执行")
        print("="*60)

        executor_response = self.agents.get("executor").process_request(
            f"请执行以下任务:{task.description}"
        )
        results["steps"].append({
            "stage": "execution",
            "agent": "executor",
            "output": executor_response
        })
        task.result = executor_response
        print(f"执行结果: {executor_response[:100]}...")

        # 步骤3: 评审
        print("\n" + "="*60)
        print("阶段3: 结果评审")
        print("="*60)

        reviewer_response = self.agents.get("reviewer").process_request(
            f"请评审以下执行结果并给出建议:\n{executor_response}"
        )
        results["steps"].append({
            "stage": "review",
            "agent": "reviewer",
            "output": reviewer_response
        })
        print(f"评审结果: {reviewer_response[:100]}...")

        # 更新任务状态
        task.status = "completed"

        return results

# 测试多代理系统
if __name__ == "__main__":
    # 创建多代理系统
    system = MultiAgentSystem()

    # 注册多个代理
    system.register_agent("planner", AgentRole.PLANNER, "小规划师")
    system.register_agent("executor", AgentRole.EXECUTOR, "小执行者")
    system.register_agent("reviewer", AgentRole.REVIEWER, "小评审员")

    print("\n" + "="*60)
    print("开始多代理协作测试")
    print("="*60 + "\n")

    # 创建一个复杂任务
    task_id = system.create_task("分析Python中的装饰器模式并提供代码示例")

    # 执行工作流程
    results = system.execute_workflow(task_id)

    # 显示最终结果
    print("\n" + "="*60)
    print("工作流程执行完成")
    print("="*60)
    print(f"任务: {results['description']}")
    print(f"执行步骤数: {len(results['steps'])}")

常见应用场景 / 实战案例分析

了解了核心功能后,让我们看看微软教程中推荐的一些常见应用场景。这些场景涵盖了AI Agent的典型用途,可以作为开发自己应用的参考。

场景一:智能客服系统

智能客服是最常见的AI Agent应用场景之一。一个好的智能客服Agent应该能够理解用户的问题、提供准确的答案、处理常见的业务流程,并在必要时转接人工客服。

在微软的教程中,推荐的实现方式包括:使用Semantic Kernel定义客服技能集,包括FAQ回答、订单查询、退换货处理等;通过对话管理维护上下文,实现多轮对话;集成知识库系统,提供准确的回复;设置升级机制,在Agent无法处理时转接人工。

典型的智能客服Agent架构包括以下几个层次:最上层是自然语言理解层,负责解析用户输入;中间是对话管理层,维护对话状态和流程;底层是技能执行层,调用具体的业务逻辑和工具。

场景二:个人助理与日程管理

AI Agent作为个人助理,可以帮助用户管理日程、回复邮件、整理信息、设置提醒等。这类应用需要Agent具备较强的任务规划能力和工具使用能力。

实现日程管理功能时,通常需要集成日历API,如Google Calendar或Outlook Calendar。Agent需要理解用户的自然语言指令,如”明天上午10点开会”,并准确地创建日历事件。高级功能还包括冲突检测、时间建议、自动提醒等。

邮件处理是另一个实用功能。Agent可以自动阅读邮件、分类优先级、生成回复草稿,甚至根据用户设置自动发送标准回复。这类功能需要特别注意隐私和安全问题。

场景三:数据分析与报告生成

AI Agent可以成为强大的数据分析助手,帮助用户处理数据、生成可视化、撰写报告。用户可以用自然语言描述想要分析的数据和关注的问题,Agent自动完成数据提取、处理、分析和呈现的全过程。

这种应用的核心挑战在于数据处理和代码生成。Agent需要能够理解数据结构、编写数据处理代码、生成分析图表、解释分析结果。微软的教程提供了使用Python进行数据分析的详细示例,包括Pandas数据处理、Matplotlib可视化、统计推断等。

场景四:代码审查与开发辅助

对于软件开发者,AI Agent可以作为智能的开发助手,帮助进行代码审查、bug修复、代码解释、测试生成等工作。这类应用需要Agent具备较强的编程能力和领域知识。

代码审查功能通常包括:检测潜在的bug和安全漏洞、评估代码质量和风格一致性、检查是否符合最佳实践、提出改进建议。Agent可以集成到CI/CD流程中,在代码提交时自动进行审查。

场景五:研究与信息汇总

AI Agent可以充当研究助手,帮助用户搜集信息、整理资料、撰写摘要、对比分析。在学术研究、市场调研、竞品分析等场景中特别有用。

这类应用需要Agent具备强大的搜索和信息处理能力。Agent需要能够从多个来源获取信息、过滤无关内容、提取关键信息、综合分析并生成报告。微软教程中包含了使用搜索API和知识库的示例。


最佳实践与优化建议

在实际开发AI Agent应用时,遵循一些最佳实践可以显著提高系统的质量和可靠性。以下是微软教程中强调的几个重要方面。

系统提示词设计

系统提示词(System Prompt)是定义Agent行为的关键。一个好的系统提示词应该清晰、准确、具体,涵盖Agent的角色定位、能力范围、行为规则和限制条件。

设计系统提示词时,应该明确Agent的身份和职责,让用户知道Agent能做什么、不能做什么。应该提供具体的例子,帮助Agent理解期望的行为模式。应该包含必要的安全指导和约束,防止Agent产生有害输出。应该说明如何处理边界情况和错误。

提示词不是一次性设计完成的,而是需要通过测试和迭代不断优化的。建议建立测试用例库,定期评估Agent在不同场景下的表现,并根据结果调整提示词。

错误处理与容错机制

健壮的错误处理是一个可靠系统的必要条件。AI Agent可能遇到的错误包括:API调用失败、响应超时、输入格式错误、工具执行异常等。对于每种可能的错误,都应该有相应的处理策略。

基本的错误处理策略包括:优雅降级,当某个功能不可用时提供替代方案;重试机制,对于临时性故障进行自动重试;用户友好的错误提示,用清晰的语言向用户解释问题;详细日志记录,便于排查问题。

"""
错误处理示例
展示如何构建健壮的错误处理机制
"""

import time
import logging
from functools import wraps
from typing import Callable, Any

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

def retry_on_failure(max_retries: int = 3, delay: float = 1.0):
    """
    重试装饰器

    参数:
        max_retries: 最大重试次数
        delay: 重试间隔(秒)
    """
    def decorator(func: Callable) -> Callable:
        @wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            last_exception = None

            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    last_exception = e
                    logger.warning(
                        f"尝试 {attempt + 1}/{max_retries} 失败: {e}"
                    )

                    if attempt < max_retries - 1:
                        time.sleep(delay)

            # 所有重试都失败
            logger.error(f"函数 {func.__name__}{max_retries} 次尝试后仍然失败")
            raise last_exception

        return wrapper
    return decorator

class AgentError(Exception):
    """Agent基础异常类"""
    pass

class ToolExecutionError(AgentError):
    """工具执行错误"""
    pass

class APIError(AgentError):
    """API调用错误"""
    pass

class ValidationError(AgentError):
    """输入验证错误"""
    pass

@retry_on_failure(max_retries=3, delay=2.0)
def call_language_model(prompt: str) -> str:
    """调用语言模型API,带重试机制"""

    # 模拟API调用
    import random

    # 模拟随机失败(10%概率)
    if random.random() < 0.1:
        raise APIError("API暂时不可用")

    return f"模型对 '{prompt}' 的响应"

def execute_with_error_handling(tool_name: str, func: Callable, *args, **kwargs):
    """
    带错误处理的工具执行函数

    参数:
        tool_name: 工具名称
        func: 要执行的函数
        *args, **kwargs: 函数参数

    返回:
        执行结果或错误信息
    """
    try:
        logger.info(f"开始执行工具: {tool_name}")
        result = func(*args, **kwargs)
        logger.info(f"工具 {tool_name} 执行成功")
        return {"success": True, "result": result}

    except ValidationError as e:
        logger.error(f"输入验证错误: {e}")
        return {
            "success": False,
            "error_type": "validation",
            "message": f"请检查输入格式:{e}"
        }

    except ToolExecutionError as e:
        logger.error(f"工具执行错误: {e}")
        return {
            "success": False,
            "error_type": "tool_execution",
            "message": f"工具执行失败:{e}"
        }

    except APIError as e:
        logger.error(f"API调用错误: {e}")
        return {
            "success": False,
            "error_type": "api",
            "message": f"服务暂时不可用,请稍后重试"
        }

    except Exception as e:
        logger.error(f"未知错误: {e}")
        return {
            "success": False,
            "error_type": "unknown",
            "message": f"发生未知错误:{e}"
        }

# 测试错误处理
if __name__ == "__main__":
    print("=== 错误处理机制测试 ===\n")

    # 测试正常执行
    result = execute_with_error_handling(
        "calculate",
        calculate,
        "10 + 20"
    )
    print(f"计算结果: {result}\n")

    # 测试输入验证错误
    result = execute_with_error_handling(
        "calculate",
        calculate,
        "10 + 20; DROP TABLE users"
    )
    print(f"验证错误测试: {result}\n")

性能优化策略

AI Agent应用的性能优化涉及多个层面。在API调用层面,可以使用缓存机制存储常见查询的结果,减少重复调用。在响应生成层面,可以实现流式输出,提供更快的首字响应时间。在并发处理层面,可以使用异步IO,同时处理多个请求。

"""
性能优化示例
展示缓存、异步处理等优化技术
"""

import time
import hashlib
import json
from functools import lru_cache
from collections import OrderedDict
from typing import Any, Optional
import asyncio

class LRUCache:
    """最近最少使用缓存"""

    def __init__(self, capacity: int = 100):
        self.cache = OrderedDict()
        self.capacity = capacity

    def get(self, key: str) -> Optional[Any]:
        """获取缓存值"""
        if key not in self.cache:
            return None

        # 移到末尾表示最近使用
        self.cache.move_to_end(key)
        return self.cache[key]

    def put(self, key: str, value: Any):
        """设置缓存值"""
        if key in self.cache:
            self.cache.move_to_end(key)
        else:
            if len(self.cache) >= self.capacity:
                # 移除最旧的项
                self.cache.popitem(last=False)

        self.cache[key] = value

    def clear(self):
        """清空缓存"""
        self.cache.clear()

class ResponseCache:
    """响应缓存管理器"""

    def __init__(self, capacity: int = 1000):
        self.cache = LRUCache(capacity)
        self.hits = 0
        self.misses = 0

    def _generate_key(self, prompt: str, context: dict = None) -> str:
        """生成缓存键"""
        content = json.dumps({"prompt": prompt, "context": context}, sort_keys=True)
        return hashlib.md5(content.encode()).hexdigest()

    def get_cached_response(self, prompt: str, context: dict = None) -> Optional[str]:
        """获取缓存的响应"""
        key = self._generate_key(prompt, context)
        result = self.cache.get(key)

        if result:
            self.hits += 1
            print(f"缓存命中!命中次数: {self.hits}")
        else:
            self.misses += 1
            print(f"缓存未命中。未命中次数: {self.misses}")

        return result

    def cache_response(self, prompt: str, response: str, context: dict = None):
        """缓存响应"""
        key = self._generate_key(prompt, context)
        self.cache.put(key, response)

    def get_stats(self) -> dict:
        """获取缓存统计"""
        total = self.hits + self.misses
        hit_rate = (self.hits / total * 100) if total > 0 else 0

        return {
            "hits": self.hits,
            "misses": self.misses,
            "hit_rate": f"{hit_rate:.2f}%"
        }

# 异步处理示例
class AsyncAgent:
    """异步AI Agent"""

    def __init__(self):
        self.semaphore = asyncio.Semaphore(5)  # 限制并发数

    async def process_request_async(self, user_id: str, prompt: str) -> dict:
        """异步处理请求"""
        async with self.semaphore:
            print(f"开始处理用户 {user_id} 的请求...")

            # 模拟异步API调用
            await asyncio.sleep(1)

            result = f"用户 {user_id} 的响应内容"

            return {
                "user_id": user_id,
                "response": result,
                "timestamp": time.time()
            }

    async def batch_process(self, requests: list) -> list:
        """批量异步处理请求"""
        tasks = [
            self.process_request_async(req["user_id"], req["prompt"])
            for req in requests
        ]

        results = await asyncio.gather(*tasks)
        return results

# 测试性能优化
if __name__ == "__main__":
    print("=== 缓存机制测试 ===\n")

    cache = ResponseCache()

    # 测试缓存
    for i in range(5):
        print(f"\n--- 请求 {i + 1} ---")
        if i < 3:
            cache.get_cached_response(f"你好,告诉我关于{i}的事情")
            cache.cache_response(f"你好,告诉我关于{i}的事情", f"这是关于{i}的信息")
        else:
            # 重复请求前3个,测试缓存命中
            cache.get_cached_response(f"你好,告诉我关于{i-3}的事情")

    print(f"\n缓存统计: {cache.get_stats()}")

    print("\n=== 异步处理测试 ===\n")

    async def test_async():
        agent = AsyncAgent()

        # 创建一批请求
        requests = [
            {"user_id": f"user_{i}", "prompt": f"请求 {i}"}
            for i in range(10)
        ]

        start_time = time.time()
        results = await agent.batch_process(requests)
        elapsed = time.time() - start_time

        print(f"\n处理了 {len(results)} 个请求")
        print(f"总耗时: {elapsed:.2f} 秒")
        print(f"平均每个请求: {elapsed/len(results):.2f} 秒")

    asyncio.run(test_async())

安全性考虑

AI Agent应用的安全性不容忽视。需要考虑的方面包括:输入验证,防止恶意输入导致安全问题;敏感信息处理,避免在日志中暴露API密钥等敏感数据;输出过滤,检测和过滤可能的有害内容;权限控制,确保Agent只能访问授权的资源。

在设计系统时,应该遵循最小权限原则,Agent只应拥有完成其任务所必需的最小权限。对于涉及敏感操作的场景,应该添加额外的确认机制。


总结与延伸学习资源

通过这篇教程,我们详细了解了微软AI Agents for Beginners项目以及AI Agent开发的核心知识。从基础概念到代码实现,从单Agent到多Agent协作,我们构建了一个功能相对完整的AI Agent系统。

核心要点回顾:

AI Agent是由感知、规划、记忆、工具使用和行动执行等多个模块组成的智能系统。Semantic Kernel和AutoGen是微软官方推荐的两个核心开发框架,各有特色,适用于不同的场景。工具调用是让Agent具有实用价值的关键能力,包括代码执行、搜索查询、API调用等。多Agent协作可以处理更复杂的任务,通过角色分工实现专业化。健壮的错误处理和性能优化是生产环境应用的必要条件。

延伸学习资源:

如果你希望进一步深入学习AI Agent开发,以下资源值得关注:

微软Semantic Kernel官方文档提供了完整的框架使用指南和大量示例代码。AutoGen项目页面有详细的框架介绍和多Agent系统的最佳实践。OpenAI的官方文档详细说明了GPT模型的API使用方法。LangChain是一个流行的AI应用开发框架,也有丰富的学习资源。Hugging Face提供了大量的开源模型和工具,适合学习prompt工程和模型微调。

微软AI Agents for Beginners项目本身也在持续更新中,建议定期关注项目仓库获取最新内容和功能。

下一步建议:

对于学完本教程的读者,可以尝试以下进阶项目:用本教程学到的知识开发一个实际的应用,如个人助手、自动化工具等;深入研究某个特定的框架,如Semantic Kernel或AutoGen,掌握其高级特性;探索多Agent系统的设计模式,学习如何构建更复杂的协作系统;尝试接入不同的语言模型和工具,比较它们的特点和适用场景;学习AI Agent的评估和优化方法,建立自己的测试和迭代流程。

AI Agent是一个快速发展的领域,新的技术和应用层出不穷。保持学习和实践的态度,你一定能在这个领域取得进步。

祝你学习愉快!

如果内容对您有帮助,欢迎打赏

您的支持是我继续创作的动力

前往打赏页面

评论区

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注