AI Agent开发入门笔记(1)
目录
- 1️⃣ 选择框架
- 2️⃣开发操作
- 导入Python库
- 创建功能函数
- 装载环境变量
- 创建Agent
- 运行Agent
学习参考资料:
- 微软 AI Agents for Beginners
- 代码仓库
1️⃣ 选择框架
semantic-kernel
开发框架
- 导入库
- 创建功能函数(Agent 要完成什么功能)
- 创建客户端
- 加载环境变量(保存在
.env,
保证API_Key
安全性) - 配置基地址(
URL
) - 选择基座模型
- 加载环境变量(保存在
- 创建Agent
- 将客户端配置到
service
- Agent 名称
- 将客户端配置到
- 运行Agent
可能头疼地方的解决方案:
- 模型API付费:
- 创建
Github Token
- 基地址:https://models.inference.ai.azure.com/
- 创建
2️⃣开发操作
导入Python库
import os
from typing import Annotated
from openai import AsyncOpenAIfrom dotenv import load_dotenvfrom semantic_kernel.agents import ChatCompletionAgent, ChatHistoryAgentThread
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion
from semantic_kernel.functions import kernel_function
创建功能函数
import random # Define a sample plugin for the sampleclass DestinationsPlugin:"""A List of Random Destinations for a vacation."""def __init__(self):# List of vacation destinationsself.destinations = ["Barcelona, Spain","Paris, France","Berlin, Germany","Tokyo, Japan","Sydney, Australia","New York, USA","Cairo, Egypt","Cape Town, South Africa","Rio de Janeiro, Brazil","Bali, Indonesia"]# Track last destination to avoid repeatsself.last_destination = None@kernel_function(description="Provides a random vacation destination.")def get_random_destination(self) -> Annotated[str, "Returns a random vacation destination."]:# Get available destinations (excluding last one if possible)available_destinations = self.destinations.copy()if self.last_destination and len(available_destinations) > 1:available_destinations.remove(self.last_destination)# Select a random destinationdestination = random.choice(available_destinations)# Update the last destinationself.last_destination = destinationreturn destination
装载环境变量
load_dotenv()
client = AsyncOpenAI(api_key=os.environ.get("GITHUB_TOKEN"), base_url="https://models.inference.ai.azure.com/",
)# Create an AI Service that will be used by the `ChatCompletionAgent`
chat_completion_service = OpenAIChatCompletion(# 模型可自行选择ai_model_id="gpt-4o-mini",async_client=client,
)
创建Agent
agent = ChatCompletionAgent(service=chat_completion_service, plugins=[DestinationsPlugin()],name="TravelAgent",instructions="You are a helpful AI Agent that can help plan vacations for customers at random destinations",
)
运行Agent
async def main():# Create a new thread for the agent# If no thread is provided, a new thread will be# created and returned with the initial responsethread: ChatHistoryAgentThread | None = Noneuser_inputs = ["Plan me a day trip.",]for user_input in user_inputs:print(f"# User: {user_input}\n")first_chunk = Trueasync for response in agent.invoke_stream(messages=user_input, thread=thread,):# 5. Print the responseif first_chunk:print(f"# {response.name}: ", end="", flush=True)first_chunk = Falseprint(f"{response}", end="", flush=True)thread = response.threadprint()# Clean up the threadawait thread.delete() if thread else Noneawait main()