30分钟快速上手AI Agent：从概念到实践构建智能体闭环系统

import os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ.get('DEEPSEEK_API_KEY'),
    base_url="https://api.deepseek.com")
response = client.chat.completions.create(
    model="deepseek-chat",
    messages=[
        {"role": "system", "content": "You are a helpful assistant"},
        {"role": "user", "content": "Hello"},
    ],
    stream=False
)
print(response.choices[0].message.content)

import json
import openai

def get_weather(destination, date):
    return f"{destination} {date} 天气晴朗"
def get_attractions(destination):
    return f"{destination} 的热门景点有故宫、颐和园..."
# ... 其他工具函数实现

# 系统提示词（内容同上）
system_prompt = "..."

messages = [
    {"role": "system", "content": system_prompt},
    {"role": "user", "content": “帮我规划下周去北京的行程“}
]

client = OpenAI(
    api_key=os.environ.get('DEEPSEEK_API_KEY'),
    base_url="https://api.deepseek.com")

while True:
    response = client.chat.completions.create(
    model="deepseek-chat",
    messages=messages,
    stream=False
    )
    # 解析 AI 返回的 JSON
    try:
        decision = json.loads(response.choices[0].message.content)
    except:
        print(“模型输出非 JSON，错误处理...“)
        break

    if decision[“action”] == “call_tool”:
        tool_name = decision[“tool”]
        params = decision[“parameters”]
        # 调用对应工具
        if tool_name == “get_weather”:
            result = get_weather(**params)
        elif tool_name == “get_attractions”:
            result = get_attractions(**params)
        # ... 其他工具调用
        # 将工具执行结果加入对话历史
        messages.append({"role": “tool”, “content”: result, “tool_call_id”: tool_name})
        # 继续循环，让AI基于新信息做下一步决策
    elif decision[“action”] == “respond”:
        print(decision[“message”])
        break

tools = [
{
“type”: “function”,
“function”: {
    “name”: “get_weather”,
    “description”: “查询目的地的天气预报”,
    “parameters”: {
        “type”: “object”,
        “properties”: {
            “destination”: {
                “type”: “string”,
                “description”: “目的地城市名称”
            },
            “date”: {
                “type”: “string”,
                “description”: “查询日期，格式为 YYYY-MM-DD，如果不指定则返回未来几天的天气”
            }
        },
        “required”: [“destination”]
    }
}
},
# 其他函数（get_attractions, get_hotels, get_route）以同样格式定义
]

system_prompt = “你是一位专业的旅游规划助手。
你的目标是根据用户的需求，
调用可用的工具获取信息，并整合成一份详尽的旅游行程建议。“

import openai
import json

# 初始化客户端（DeepSeek API 兼容 OpenAI 格式）
client = openai.OpenAI(
    api_key=“your-deepseek-api-key”,  # 替换为你的 API Key
    base_url=“https://api.deepseek.com”
)
# 工具的具体实现（模拟）
def get_weather(destination, date=None):
    # 实际场景中应调用天气 API
    return f“{destination} 的天气晴朗，气温15-25℃，适合出行。”
# ... 其他工具函数省略

messages = [
    {“role”: “system”, “content”: “你是一个专业的旅游规划助手”},
    {“role”: “user”, “content”: user_input}
]

while True:
    result = client.chat.completions.create(
      messages=messages,
      model=“deepseek-chat”,
      stream=False,
      tools=tools) # 将定义好的tools传入
    assistant_msg = result.choices[0].message
    messages.append(assistant_msg)

    # 检查模型是否要求调用工具
    if assistant_msg.tool_calls:
        for tool_call in assistant_msg.tool_calls:
            function_name = tool_call.function.name
            function_args = json.loads(tool_call.function.arguments)

            # 调用对应的函数
            if function_name == “get_weather”:
                tool_result = get_weather(**function_args)
            elif function_name == “get_attractions”:
                tool_result = get_attractions(**function_args)
            # ... 其他工具

            # 将工具执行结果以特定格式加入对话历史
            messages.append({
                “role”: “tool”,
                “content”: str(tool_result),
                “tool_call_id”: tool_call.id
            })
        # 继续循环，让模型基于工具结果进行下一步推理
    else:
        # 模型返回了最终答案
        final_answer = assistant_msg.content
        print(f“助手：{final_answer}”)
        break

from collections import deque

class ShortTermMemory:
    def __init__(self, max_messages=20):
        “”“
        初始化短期记忆。
        :param max_messages: 最多保留的消息条数。
        ”“”
        self.messages = deque(maxlen=max_messages)

    def add(self, message: dict):
        “”“添加一条消息到短期记忆。”“”
        self.messages.append(message)

    def get_all(self) -> list:
        “”“返回当前所有消息。”“”
        return list(self.messages)

    def clear(self):
        self.messages.clear()

import numpy as np
from typing import List

class LongTermMemory:
    def __init__(self, embedding_model):
        self.embedding_model = embedding_model
        self.vectors = []  # 存储向量
        self.texts = []     # 存储原始文本

    def _embed(self, text: str) -> np.ndarray:
        return self.embedding_model.encode(text) # 假设有embedding模型

    def add(self, text: str, metadata: dict = None):
        “”“将文本存入长期记忆。”“”
        vector = self._embed(text)
        self.vectors.append(vector)
        self.texts.append(text)

    def query(self, query_text: str, top_k: int = 3) -> List[str]:
        “”“根据查询文本，返回最相关的 top_k 条记忆。”“”
        if not self.vectors:
            return []
        query_vec = self._embed(query_text)
        # 计算余弦相似度 (简化)
        similarities = [
            np.dot(query_vec, v) / (np.linalg.norm(query_vec) * np.linalg.norm(v))
            for v in self.vectors
        ]
        top_indices = np.argsort(similarities)[-top_k:][::-1]
        return [self.texts[i] for i in top_indices]

class MemoryManager:
    def __init__(self, embedding_model, short_term_max_messages=20, long_term_top_k=3):
        self.short_term = ShortTermMemory(max_messages=short_term_max_messages)
        self.long_term = LongTermMemory(embedding_model)
        self.long_term_top_k = long_term_top_k

    def add_short_term(self, message: dict):
        self.short_term.add(message)

    def add_long_term(self, text: str, metadata: dict = None):
        # 实际应用中可在此处进行重要性过滤
        self.long_term.add(text, metadata)

    def build_messages(self, current_query: str, system_prompt_base: str = “”) -> List[dict]:
        # 1. 从长期记忆检索相关背景
        long_memories = self.long_term.query(current_query, top_k=self.long_term_top_k)
        # 2. 构建增强的系统提示词
        if long_memories:
            memory_context = “以下是可能与当前问题相关的历史记忆：\n” + “\n”.join(f”- {mem}” for mem in long_memories)
            system_content = f“{system_prompt_base}\n\n{memory_context}”
        else:
            system_content = system_prompt_base
        # 3. 获取短期记忆（最近对话）
        short_context = self.short_term.get_all()
        # 4. 组合所有部分
        messages = [{“role”: “system”, “content”: system_content}]
        messages.extend(short_context)
        messages.append({“role”: “user”, “content”: current_query})
        return messages

# 初始化记忆管理器（需传入一个embedding模型实例）
memory = MemoryManager(embedding_model=some_embedding_model)
system_base = “你是一位专业的旅游规划助手。你可以调用工具获取天气、景点、酒店和路线信息。”

while True:
    user_input = input(“用户：”)
    if user_input.lower() in (“exit”, “quit”):
        break

    # 关键步骤：构建带有记忆的对话上下文
    messages = memory.build_messages(user_input, system_prompt_base=system_base)

    # 原有的工具调用循环（此处省略，与之前Function Calling部分相同）
    # ... 调用模型，处理工具调用，得到最终答案 final_answer ...

    # 将本轮完整的交互存入短期记忆
    memory.add_short_term({“role”: “user”, “content”: user_input})
    memory.add_short_term({“role”: “assistant”, “content”: final_answer})

    # 示例：将用户明确表达的偏好或最终计划存入长期记忆
    if “我喜欢” in user_input or “偏好” in user_input:
        memory.add_long_term(f“用户偏好：{user_input}”, metadata={“type”: “preference”})
    if “规划” in user_input and final_answer:
        memory.add_long_term(f“生成的旅游计划摘要：{final_answer[:100]}...”, metadata={“type”: “plan”})

    print(f“助手：{final_answer}”)