Claude智能体开发入门指南：从零构建你的第一个AI助手

1次阅读

共计 3490 个字符，预计需要花费 9 分钟才能阅读完成。

AI 智能体（Agent）可以理解为具备特定领域对话能力的程序。它能理解用户输入（文本 / 语音），通过预设逻辑或机器学习模型生成响应。核心特点包括：

意图识别：判断用户说话目的（如 ” 查天气 ” 对应天气查询功能）
上下文管理：记住对话历史（如用户先说 ” 北京 ” 再问 ” 天气怎么样 ”）
多轮对话：通过多个对话轮次（turn）完成复杂任务

客服机器人（处理退货、查询订单）
智能家居控制（” 打开客厅的灯 ”）
教育辅导（数学题分步讲解）

方式	优点	缺点
官方 SDK	封装完善，开发速度快	灵活性较低
原生 API 调用	完全控制请求流程	需自行处理签名、重试等

Python 3.10+（含 pip）
安装必要库：

pip install requests python-dotenv

获取 API Key：登录 Claude 开发者平台创建应用

创建claude_client.py：

import os
import requests
from dotenv import load_dotenv

load_dotenv()  # 加载.env 中的 API_KEY

class ClaudeClient:
    def __init__(self):
        self.api_key = os.getenv("CLAUDE_API_KEY")
        self.base_url = "https://api.claude.ai/v1"

    def send_message(self, text, session_id=None):
        headers = {"Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }

        payload = {"text": text}
        if session_id:
            payload["session_id"] = session_id

        try:
            response = requests.post(f"{self.base_url}/chat",
                json=payload,
                headers=headers,
                timeout=10
            )
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            print(f"API 请求失败: {e}")
            return None

创建dialogue_manager.py：

from collections import defaultdict

class DialogueManager:
    def __init__(self):
        self.sessions = defaultdict(dict)  # 存储会话状态

    def handle_message(self, user_id, text):
        # 获取或初始化会话
        session = self.sessions[user_id]

        # 基础上下文保持（最后 5 轮对话）context = session.get("context", [])
        context.append(text)
        if len(context) > 5:
            context = context[-5:]
        session["context"] = context

        # 调用 API（示例简化处理）client = ClaudeClient()
        response = client.send_message(text, session_id=user_id)

        # 更新会话状态
        if response:
            session["last_response"] = response["text"]
            return response["text"]
        return "抱歉，服务暂时不可用"

在 DialogueManager 类中添加：

import re

def _detect_intent(self, text):
    # 简单正则匹配
    patterns = {"greeting": r"(你好 | 嗨 |hello)",
        "weather": r"(天气 | 下雨 | 气温)",
        "goodbye": r"(再见 | 拜拜 |exit)"
    }

    for intent, pattern in patterns.items():
        if re.search(pattern, text, re.IGNORECASE):
            return intent
    return "unknown"

修改 ClaudeClient.send_message 方法：

from time import sleep

def send_message(self, text, session_id=None, max_retries=3):
    # ...（原有 headers 和 payload 代码）for attempt in range(max_retries):
        try:
            # 指数退避重试
            if attempt > 0:
                sleep(2 ** attempt)

            response = requests.post(...)

            # 处理速率限制
            if response.status_code == 429:
                retry_after = int(response.headers.get("Retry-After", 1))
                sleep(retry_after)
                continue

            response.raise_for_status()
            return response.json()

        except requests.exceptions.RequestException as e:
            if attempt == max_retries - 1:
                print(f"最终失败: {e}")
                return None

创建safety_filter.py：

class SafetyFilter:
    BANNED_WORDS = ["密码", "银行卡", "身份证"]

    @classmethod
    def filter(cls, text):
        for word in cls.BANNED_WORDS:
            if word in text:
                return "您输入的内容包含敏感信息，已屏蔽"
        return text

建议使用 Logstash 或直接写入数据库：

import json
from datetime import datetime

class DialogueLogger:
    def log(self, user_id, request, response):
        entry = {"timestamp": datetime.utcnow().isoformat(),
            "user_id": user_id,
            "request": request,
            "response": response,
            "context": self.sessions[user_id].get("context", [])
        }

        # 写入文件（生产环境建议用 ELK 或数据库）with open("dialogue.log", "a") as f:
            f.write(json.dumps(entry) + "\n")

错误表现：UI 卡死等待响应

解决方案：