1. 从“玩具”到“生产力”为什么现在必须掌握智能体开发如果你最近在技术社区里泡着大概率会频繁刷到“AI智能体”这个词。从年初的AutoGPT、BabyAGI到后来各大模型厂商推出的“GPTs”、“Copilot Studio”再到如今Claude、GPT-4o等模型原生支持的“Code Skills”或“Function Calling”能力智能体已经从实验室里的概念变成了触手可及的生产力工具。很多人可能觉得这不就是让AI调用几个API吗有什么大不了的我最初也是这么想的直到我亲手用Claude Code Skills在几个小时内把一个需要反复手动查询、复制粘贴数据的日报生成流程自动化成了一个能自主运行、定时汇报的Bot我才意识到这件事的本质变了。过去我们让AI写代码、回答问题它更像一个“超级实习生”你给指令它产出内容但执行和串联的“手”和“脑”还是你自己。而智能体是让AI自己拥有了“手”和“眼睛”。这个“手”就是Code Skills——模型直接编写并执行代码的能力。这意味着AI不再只是给你一段需要你手动去跑的Python脚本而是可以自己思考“要完成用户‘获取今日销售额并分析趋势’这个任务我需要先调用公司CRM的API技能A拿到数据后再用matplotlib画个图技能B最后把图表和总结发到Slack频道技能C。” 然后它自己就把这一套流程给执行了。所以“从零开发一个Bot智能体”在今天已经不是一个炫技的玩具项目而是一个实实在在的效能倍增器。它解决的核心痛点是将复杂、多步骤、跨工具的手动操作封装成一个由自然语言驱动的、可重复执行的自动化流程。无论是每天早上的数据拉取与简报、监控告警的自动分析与初步排查还是跨系统信息的查询与整合一个设计良好的Bot智能体都能让你从重复劳动中彻底解放出来。Claude的Code Skills在这一波浪潮中显得尤为突出因为它背后是Claude 3系列模型强大的代码生成与推理能力。与简单的函数调用不同Code Skills允许模型在一个持久的、安全的沙箱环境中动态生成并运行代码处理文件安装临时依赖。这相当于给了AI一个功能完整的“临时工作台”其灵活性和解决问题的能力上了一个大台阶。接下来我就以打造一个“项目进度追踪与自动汇报Bot”为例带你完整走一遍实战流程分享从环境准备、技能设计、到调试部署的每一个细节以及我趟过的那些坑。2. 战场准备理解Claude Code Skills的运行机制与边界在撸起袖子写第一个技能之前我们必须先摸清楚手中的武器——Claude Code Skills——到底是怎么工作的以及它的能力边界在哪里。这决定了我们设计智能体的思路避免后期出现“我以为它能结果它不能”的尴尬局面。2.1 核心机制沙箱、会话与技能库Claude Code Skills不是让AI模型在你的本地终端或服务器上为所欲为。它的核心是一个受控的代码执行沙箱环境。当你通过API或Claude Console触发一个Code Skills调用时会发生以下几步意图识别Claude模型首先理解你的自然语言请求例如“帮我看看上周项目日志里ERROR级别的记录有多少条”。技能匹配与规划模型会判断是否需要以及如何使用已配置的技能。它会思考“用户需要分析日志文件。我有一个‘文件读取’技能可以读取指定路径的文件内容。我还需要一个‘文本分析’技能来统计ERROR关键词。我需要先调用技能A将其结果作为输入传递给技能B。”代码生成与执行对于需要Code Skills的任务模型会在沙箱中动态生成Python代码这是目前的主要语言。这个沙箱是临时的、隔离的通常具备基础Python环境、网络访问能力可配置和有限的临时磁盘空间。代码在此沙箱中运行。结果返回代码执行的标准输出stdout、错误信息stderr以及可能的文件产出会作为结果返回给Claude模型。模型再对这些结果进行解读、总结并用自然语言回复给你。这里的关键在于“技能”Skills的配置。你可以预先定义好一系列技能每个技能本质上是一个允许模型执行的操作的抽象描述。例如一个“HTTP请求”技能告诉模型“你可以向互联网发送GET/POST请求来获取数据”。一个“数据库查询”技能告诉模型“你可以连接到一个安全的数据库通过预配置的连接信息执行SQL查询”。一个“运行Shell命令”技能告诉模型“你可以在沙箱中执行特定的、允许的系统命令”。模型根据这些技能描述来决定如何组合它们以完成任务。Code Skills则是更底层、更灵活的一环当预定义的技能不够用时模型可以直接编写代码来实现复杂逻辑。2.2 能力边界与安全红线理解边界比理解能力更重要这直接关系到项目的可行性与安全性。持久化与状态每次Code Skills调用沙箱环境通常是全新的。这意味着你不能指望在一次会话中在/tmp目录下写一个文件然后在下次调用中还能读到它。智能体的“状态”需要通过外部存储来维护比如数据库、文件存储服务S3等或你在技能中提供的API。执行时间与资源沙箱有执行时间限制例如30秒或60秒和内存/CPU限制。长时间运行的后台任务、需要大量计算的任务如训练机器学习模型不适合在此运行。网络访问沙箱可以访问外网但这通常需要显式启用或配置。出于安全考虑某些内部网络地址或端口可能会被阻止。文件系统访问沙箱对文件系统的访问是受限的通常只能访问临时目录或你通过技能特别挂载的路径。它不能随意读写宿主机上的任意文件。安全性最重要这是双刃剑。沙箱保护了你的主机环境但你也必须警惕提示词注入避免让用户输入直接成为代码或命令的一部分。永远要对用户提供的参数进行严格的校验和清理。敏感信息泄露绝对不要将API密钥、数据库密码等硬编码在技能描述或可能被模型生成的代码中。应该使用环境变量或安全的密钥管理服务并通过技能配置以安全的方式传递给沙箱环境。无限循环与资源耗尽模型生成的代码可能有bug比如死循环。沙箱通常有看门狗机制但设计技能时也要有意识避免此类模式。提示在开发初期一个非常实用的做法是在技能描述中明确写出约束条件。例如在“执行数据分析”的技能描述里加上“注意沙箱环境最多使用1GB内存运行时间不超过30秒。请确保生成的Pandas代码不会尝试读取超过100MB的文件。”2.3 我们的Bot智能体蓝图基于以上理解我们来规划要打造的“项目进度自动汇报Bot”的核心功能与架构核心目标每天上午10点自动收集指定项目的关键数据如Git提交记录、JIRA问题状态、CI/CD流水线结果生成一份结构化的进度报告并发送到团队Slack频道。架构设计触发器一个定时任务例如使用云函数、cron job或Make.com/Zapier等自动化平台定期调用我们的Bot智能体。智能体核心一个封装了Claude Code Skills的程序。它接收触发由Claude模型理解任务并协调执行一系列技能。技能集技能AGit仓库查询。通过GitHub/GitLab API获取昨日提交列表、分支合并情况。技能B项目管理JIRA数据获取。查询指定JIRA看板获取“进行中”、“待测试”、“已完成”状态的任务列表。技能CCI/CD状态检查。调用Jenkins/GitLab CI API获取最新构建的成功/失败状态。技能D数据分析与报告生成。将A、B、C技能获取的原始数据进行聚合、分析如计算完成度、阻塞项并格式化为Markdown或HTML报告。技能E消息推送。将生成的报告发送到Slack指定频道。状态/缓存层一个简单的数据库如SQLite或键值存储如Redis用于记录上次运行的时间、已处理的数据ID避免报告重复内容。这个架构将利用Code Skills的灵活性来处理数据分析和报告生成技能D而其他相对固定的API调用则可以用预定义的HTTP技能或封装好的函数来实现。3. 搭建开发环境从API Key到第一个“Hello World”技能理论说再多不如动手跑通第一个例子。这里我选择使用Anthropic官方提供的Python SDK进行开发因为它最直接也便于我们理解底层交互。3.1 获取必要的密钥与工具Anthropic API Key前往 Anthropic Console 注册并创建API Key。这是调用Claude模型的通行证。妥善保管不要提交到代码仓库。Python环境建议使用Python 3.9。使用venv或conda创建一个干净的虚拟环境。python -m venv claude-bot-env source claude-bot-env/bin/activate # Linux/macOS # claude-bot-env\Scripts\activate # Windows安装SDK在虚拟环境中安装Anthropic Python SDK。pip install anthropic3.2 编写第一个“代码执行”对话我们先不急着定义复杂的技能而是直接让Claude在沙箱里运行一段代码感受一下Code Skills最原始的能力。创建一个文件first_skill.pyimport anthropic import os # 从环境变量读取API Key更安全 client anthropic.Anthropic( api_keyos.environ.get(ANTHROPIC_API_KEY) ) # 构建一个要求Claude执行代码的对话 message client.messages.create( modelclaude-3-5-sonnet-20241022, # 使用支持Code Skills的最新模型 max_tokens1000, system你是一个有帮助的AI助手可以使用Code Skills执行Python代码来解决问题。, messages[ { role: user, content: 请帮我计算一下斐波那契数列的前10个数字并用Python代码实现告诉我结果。 } ] ) print(message.content[0].text)运行这个脚本记得先设置环境变量ANTHROPIC_API_KEY你会看到Claude不仅给出了斐波那契数列的代码还输出了执行结果[0, 1, 1, 2, 3, 5, 8, 13, 21, 34]。这说明模型在背后已经生成了代码并在沙箱中执行完毕。但这还不够“技能化”。我们更希望模型能调用我们预先定义好的、更安全可控的功能。3.3 定义并集成第一个自定义技能文件内容读取让我们定义一个更实用的技能读取指定项目目录下的CHANGELOG.md文件并总结最新版本更新。我们将通过SDK的tools参数来定义这个技能。首先我们按照Anthropic的tools格式定义一个技能。这本质上是一个符合OpenAI Function Calling格式的JSON Schema描述。import anthropic import os client anthropic.Anthropic(api_keyos.environ.get(ANTHROPIC_API_KEY)) # 1. 定义“读取文件”技能 read_file_tool { name: read_project_changelog, description: 读取项目根目录下的CHANGELOG.md文件内容用于分析最近的版本更新。, input_schema: { type: object, properties: {} # 这个技能不需要输入参数固定读取一个文件 required: [] } } # 2. 在实际对话中当模型决定调用此技能时我们需要一个处理函数来执行真正的逻辑 def handle_read_changelog(): 模拟读取CHANGELOG文件的处理函数 # 这里应该是真实的文件读取逻辑。例如 # with open(/path/to/project/CHANGELOG.md, r) as f: # content f.read() # 为了演示我们返回模拟数据 mock_changelog # 项目变更日志 ## [v1.2.0] - 2024-04-15 ### 新增 - 实现了用户权限管理模块。 - 添加了项目导出为PDF功能。 ## [v1.1.0] - 2024-04-01 ### 修复 - 解决了登录页面在移动端的布局错乱问题。 - 修复了数据导出的时间戳错误。 return {content: mock_changelog} # 3. 发起对话并提供技能定义 message client.messages.create( modelclaude-3-5-sonnet-20241022, max_tokens1000, system你是一个项目助理Bot。你可以使用工具来获取项目信息。, tools[read_file_tool], # 关键在这里传入技能定义 messages[ { role: user, content: 请告诉我这个项目最近一个版本更新了什么内容 } ] ) # 4. 检查模型的回复 initial_response message.content[0].text print(Claude的初始回复可能要求调用技能: , initial_response) # 在实际的流式或异步交互中模型可能会返回一个tool_use块表示它想调用某个技能。 # 我们需要检查响应如果模型要求调用工具则执行对应的处理函数并将结果追加到对话中再次请求模型。 # 下面是一个简化的同步示例逻辑 if hasattr(message, tool_calls) and message.tool_calls: # 注意实际API响应结构可能不同此处为逻辑示意 # 假设模型决定调用 read_project_changelog tool_result handle_read_changelog() # 然后将工具执行结果作为新的消息内容再次发送给模型进行总结 second_message client.messages.create( modelclaude-3-5-sonnet-20241022, max_tokens1000, system你是一个项目助理Bot。, tools[read_file_tool], messages[ {role: user, content: 请告诉我这个项目最近一个版本更新了什么内容}, {role: assistant, content: f我需要调用read_project_changelog技能来获取信息。}, # 模拟助理的请求 { role: user, # 注意工具执行结果通常以tool角色发送 content: [{ type: tool_result, tool_use_id: message.tool_calls[0].id, # 需要匹配tool call的ID content: tool_result[content] }] } ] ) print(\nClaude在获取文件内容后的总结\n, second_message.content[0].text) else: print(\nClaude直接给出了回答\n, initial_response)注意上述代码中关于tool_calls的处理是一个逻辑示意。Anthropic Messages API对于工具调用的实际处理是流式streaming的或者需要在收到tool_use区块后手动构造后续请求。最新的SDK和API文档提供了更优雅的tools集成方式。这里的目的是展示“定义技能”和“模型决定使用技能”的基本交互模式。在实际开发中你需要根据官方指南实现完整的工具调用循环。通过这一步我们明确了技能开发的基本模式定义技能描述 - 模型在对话中请求调用 - 我们的后端执行实际逻辑 - 将结果返回给模型 - 模型生成最终回复。接下来我们就要用这个模式构建我们Bot的核心技能集。4. 构建核心技能集让Bot拥有“手”和“眼睛”我们的项目进度Bot需要与多个外部系统交互。为每个系统封装一个清晰的技能是智能体稳定可靠的基础。这里我以GitHub API和Slack Incoming Webhook为例展示两个最关键技能的完整实现。4.1 技能一GitHub提交记录获取器这个技能的目标是根据给定的仓库名、日期范围获取相关的提交记录。首先我们需要一个安全的密钥管理方式。我们将使用python-dotenv来管理环境变量。安装依赖并配置环境pip install python-dotenv requests在项目根目录创建.env文件ANTHROPIC_API_KEYsk-ant-... GITHUB_TOKENghp_... # 需要repo权限的GitHub Personal Access Token GITHUB_OWNERyour_company_or_username.env文件务必加入.gitignore避免密钥泄露。实现GitHub API交互函数 创建一个模块github_integration.pyimport os import requests from datetime import datetime, timedelta from dotenv import load_dotenv load_dotenv() GITHUB_API https://api.github.com HEADERS { Authorization: ftoken {os.getenv(GITHUB_TOKEN)}, Accept: application/vnd.github.v3json } def get_recent_commits(repo_name: str, since_days: int 1) - list: 获取指定仓库最近N天内的提交记录。 Args: repo_name: 仓库名称不含所有者。 since_days: 查询多少天内的数据默认1天昨日至今。 Returns: 一个包含提交信息的字典列表。 since_time (datetime.now() - timedelta(dayssince_days)).isoformat() url f{GITHUB_API}/repos/{os.getenv(GITHUB_OWNER)}/{repo_name}/commits params {since: since_time, per_page: 50} # 限制每页数量 try: response requests.get(url, headersHEADERS, paramsparams) response.raise_for_status() # 如果状态码不是200抛出异常 commits response.json() # 简化返回数据只提取关键信息 simplified_commits [] for commit in commits: commit_info commit.get(commit, {}) author commit_info.get(author, {}) simplified_commits.append({ sha: commit.get(sha, )[:7], message: commit_info.get(message, ).split(\n)[0], # 取第一行作为概要 author: author.get(name, N/A), date: author.get(date, ), url: commit.get(html_url, ) }) return simplified_commits except requests.exceptions.RequestException as e: print(fError fetching commits from GitHub: {e}) return [] except ValueError as e: print(fError parsing GitHub response: {e}) return []将函数封装为Claude可调用的技能 在定义给Claude的tools列表中我们需要用JSON Schema来描述这个技能。# 在 main_bot.py 或类似主逻辑文件中 github_tool { name: get_github_commits, description: 获取指定GitHub仓库在最近一段时间内的提交记录。用于追踪代码变更活动。, input_schema: { type: object, properties: { repo_name: { type: string, description: GitHub仓库的名称例如my-web-app }, since_days: { type: integer, description: 查询从多少天前到现在默认是1即昨天到今天, default: 1 } }, required: [repo_name] # repo_name是必填参数 } } # 对应的处理函数 def handle_get_github_commits(repo_name: str, since_days: int 1): from github_integration import get_recent_commits # 实际项目中注意导入 commits get_recent_commits(repo_name, since_days) # 返回给Claude的结果需要是字符串或可序列化结构 if commits: # 可以格式化一下方便模型阅读 result_lines [f- {c[sha]}: {c[message]} by {c[author]} for c in commits[:10]] # 只展示前10条 return {commits: commits, summary: f共找到{len(commits)}条提交。最近几条\n \n.join(result_lines)} else: return {commits: [], summary: 在指定时间内未找到提交记录或查询出错。}现在当Claude模型认为需要获取GitHub提交信息时它会生成一个类似{repo_name: my-web-app, since_days: 1}的调用请求。我们的后端程序捕获到这个请求执行handle_get_github_commits函数并将结果返回给模型。4.2 技能二Slack消息推送器Bot生成报告后需要能推送到指定频道。我们使用Slack的Incoming Webhooks它比Bot Token更简单适合单向发送消息。在Slack创建Incoming Webhook访问 api.slack.com/apps 。创建新App或选择现有App在功能列表中找到“Incoming Webhooks”并激活。添加一个新的Webhook到你的目标频道复制生成的Webhook URL。将Webhook URL加入环境变量 在.env文件中添加SLACK_WEBHOOK_URLhttps://hooks.slack.com/services/...实现Slack推送函数 创建slack_integration.pyimport os import json import requests from dotenv import load_dotenv load_dotenv() def send_slack_message(text: str, blocks: list None, channel: str None): 通过Incoming Webhook发送消息到Slack。 Args: text: 消息的fallback文本。 blocks: Slack Block Kit格式的消息块用于富文本格式。 channel: 可覆盖默认频道的目标频道如#general。需Webhook已授权。 webhook_url os.getenv(SLACK_WEBHOOK_URL) if not webhook_url: print(错误未配置SLACK_WEBHOOK_URL环境变量。) return False payload {text: text} if blocks: payload[blocks] blocks if channel: payload[channel] channel headers {Content-Type: application/json} try: response requests.post(webhook_url, datajson.dumps(payload), headersheaders) response.raise_for_status() print(f消息已成功发送至Slack。) return True except requests.exceptions.RequestException as e: print(f发送消息到Slack失败: {e}) return False封装为Claude技能slack_tool { name: send_slack_report, description: 将格式化好的项目进度报告发送到指定的Slack频道。, input_schema: { type: object, properties: { report_markdown: { type: string, description: 完整的项目进度报告使用Markdown格式。 }, channel_override: { type: string, description: 可选指定发送到哪个Slack频道例如#project-updates。不填则使用Webhook默认频道。 } }, required: [report_markdown] } } def handle_send_slack_report(report_markdown: str, channel_override: str None): from slack_integration import send_slack_message # 将Markdown转换为Slack的mrkdwn格式Slack基本支持Markdown # 也可以使用更复杂的Block Kit来美化消息 blocks [ { type: section, text: { type: mrkdwn, text: f* 项目每日进度报告*\n{report_markdown} } } ] success send_slack_message( textf项目每日进度报告{report_markdown[:100]}..., # fallback文本 blocksblocks, channelchannel_override ) return {success: success, message: 报告已尝试发送至Slack。}按照同样的模式我们可以继续添加JIRA查询、CI状态检查等技能。关键在于每个技能的描述description要清晰准确输入参数input_schema要定义明确。这相当于给Claude模型一本清晰的“工具说明书”它才能正确地决定在什么情况下使用哪把“扳手”。5. 组装与调度构建智能体的“大脑”与工作流有了一个个独立的技能我们现在需要创建一个“大脑”——也就是主控程序来协调Claude模型和这些技能并设计一个触发它工作的机制。5.1 实现智能体主循环智能体的核心是一个循环接收用户输入或定时触发- 调用Claude模型并告知可用技能 - 解析模型响应 - 若模型要求调用技能则执行并返回结果 - 将结果反馈给模型继续对话直到模型给出最终答案。下面是一个简化但完整的主循环示例bot_agent.pyimport anthropic import os import json from dotenv import load_dotenv from typing import Dict, Any, List # 导入技能处理函数 from my_skills.github_skill import handle_get_github_commits from my_skills.slack_skill import handle_send_slack_report # ... 导入其他技能处理函数 load_dotenv() class ProjectReportBot: def __init__(self): self.client anthropic.Anthropic(api_keyos.getenv(ANTHROPIC_API_KEY)) # 定义所有可用的技能工具 self.tools [ { # GitHub技能定义 name: get_github_commits, description: 获取指定GitHub仓库在最近一段时间内的提交记录。用于追踪代码变更活动。, input_schema: { type: object, properties: { repo_name: {type: string, description: GitHub仓库名称}, since_days: {type: integer, description: 查询天数, default: 1} }, required: [repo_name] } }, { # Slack技能定义 name: send_slack_report, description: 将格式化好的项目进度报告发送到指定的Slack频道。, input_schema: { type: object, properties: { report_markdown: {type: string, description: Markdown格式报告}, channel_override: {type: string, description: 可选频道} }, required: [report_markdown] } } # ... 添加其他工具定义 ] # 映射工具名到处理函数 self.tool_handlers { get_github_commits: handle_get_github_commits, send_slack_report: handle_send_slack_report, # ... 映射其他工具 } self.conversation_history [] # 用于维护对话上下文 def _execute_tool(self, tool_name: str, tool_input: Dict[str, Any]) - Dict[str, Any]: 执行具体的工具/技能 handler self.tool_handlers.get(tool_name) if not handler: return {error: f未知的工具: {tool_name}} try: # 将字典参数解包传递给处理函数 return handler(**tool_input) except Exception as e: return {error: f执行工具{tool_name}时出错: {str(e)}} def run(self, user_query: str) - str: 运行智能体处理一次用户查询 # 1. 将用户查询加入历史 self.conversation_history.append({role: user, content: user_query}) # 2. 准备发送给Claude的消息包含全部历史 message_payload { model: claude-3-5-sonnet-20241022, max_tokens: 4000, system: 你是一个高效的项目进度助理Bot。你的任务是整合来自GitHub、JIRA等多个来源的信息生成清晰、简洁的每日项目进度报告并可以应要求将其发送到Slack。请充分利用我提供给你的工具。, tools: self.tools, messages: self.conversation_history } # 3. 发送请求并处理流式响应以处理工具调用 with self.client.messages.stream(**message_payload) as stream: final_text for event in stream: # 处理文本流 if event.type content_block_delta and event.delta.type text_delta: final_text event.delta.text # 处理工具调用请求这是关键部分 elif event.type content_block_start and event.content_block.type tool_use: tool_use_block event.content_block tool_name tool_use_block.name tool_input tool_use_block.input print(f[Bot] Claude请求调用工具: {tool_name} 参数: {tool_input}) # 4. 执行工具 tool_result self._execute_tool(tool_name, tool_input) # 将结果转换为字符串格式方便Claude读取 result_content json.dumps(tool_result, ensure_asciiFalse, indent2) print(f[Bot] 工具执行结果: {result_content[:200]}...) # 5. 将工具执行结果作为新的消息块发送回去 stream.send({ type: tool_result, tool_use_id: tool_use_block.id, content: result_content }) # 6. 将Claude的最终回复加入历史维持上下文可选取决于是否需要多轮对话 if final_text: self.conversation_history.append({role: assistant, content: final_text}) return final_text # 使用示例 if __name__ __main__: bot ProjectReportBot() # 模拟一个复杂的用户请求 query 请生成一份关于my-web-app仓库的昨日项目进度报告。 报告需要包含 1. 昨天所有的代码提交概要。 2. 如果提交数量大于5请分析一下开发活跃度。 3. 最后将这份报告用Markdown格式整理好并发送到Slack的#daily-standup频道。 report bot.run(query) print(\n *50) print(最终生成的报告) print(report)这个run方法利用Anthropic SDK的流式接口能够实时处理模型中途发起的工具调用请求形成一个完整的“思考-行动-观察”循环。5.2 设计定时触发与部署我们的Bot需要自动运行而不是手动执行脚本。这里有几个经典的部署方案方案A云函数/Serverless推荐平台Vercel Serverless Functions AWS Lambda Google Cloud Functions。流程将上述ProjectReportBot类打包。创建一个HTTP端点如/api/generate-daily-report。使用云服务商的定时触发器如AWS EventBridge Google Cloud Scheduler Vercel Cron Jobs每天定点调用这个端点。优点无需管理服务器按需付费伸缩性好。注意需要将环境变量API Keys安全地配置在云平台中。方案B传统服务器 Cron Job流程在一台长期运行的服务器或容器上部署Python脚本。使用Linux的crontab设置定时任务例如每天上午10点运行0 10 * * * /path/to/venv/bin/python /path/to/bot_agent.py /var/log/bot.log 21。优点控制力强调试方便。缺点需要维护服务器。方案C低代码自动化平台平台Make.com, Zapier, n8n。流程在这些平台上设置一个定时触发器触发一个HTTP请求Webhook到你的Bot服务端点可以是方案A的云函数也可以是一个简单的服务器。或者如果逻辑不复杂可以直接在这些平台上用内置模块调用Claude API并处理返回结果。优点配置可视化适合非开发人员或快速原型。我个人的选择是方案AVercel Serverless Cron Job。将核心逻辑部署为Serverless Function通过Vercel提供的Cron Jobs来触发几乎零运维成本极低。你需要做的就是把代码推送到GitHub并连接Vercel项目。6. 避坑指南与效能优化来自实战的经验之谈在开发和运行这个Bot的过程中我踩过不少坑也总结出一些让智能体更可靠、更高效的经验。6.1 技能设计中的三个常见陷阱技能描述过于模糊或宽泛反面例子处理数据。这种描述让模型无法准确判断何时使用。正面例子查询指定GitHub仓库在特定时间范围内的提交记录并返回提交哈希、作者、信息和时间。清晰描述了功能、输入和输出。技巧在描述中明确技能的目的、输入格式和输出格式。甚至可以加入使用示例或约束条件例如“此技能用于获取昨日至今的提交输入参数repo_name必须是字符串返回一个列表。”技能返回数据格式混乱问题处理函数返回一个复杂的Python对象如包含datetime对象的字典直接json.dumps可能会失败。或者返回的字符串过于冗长干扰模型的总结。解决在返回给模型前对数据进行序列化和精简。确保所有数据都是JSON可序列化的字符串、数字、列表、字典。提取最关键的信息过滤掉无关的元数据。例如GitHub API返回的提交信息很多我们只提取了sha、message、author、date这几个字段。缺乏错误处理与降级方案场景GitHub API临时不可用或者JIRA查询超时。如果技能直接抛出异常整个Bot流程就会中断。解决在每个技能处理函数内部做好健壮的错误处理try-catch。即使出错也返回一个结构化的错误信息而不是让程序崩溃。例如return {error: GitHub API请求失败状态码XXX, data: []}。这样模型还能在最终报告里提及“GitHub数据暂时无法获取”而不是完全卡住。6.2 提示词Prompt工程引导模型做出最佳决策系统提示词system是智能体的“人格”和“工作指南”写得好坏直接影响输出质量。基础版不够好“你是一个助手。”优化版针对我们的Bot你是一个专注、高效、严谨的项目进度助理Bot。你的核心任务是整合来自多个源头GitHub, JIRA的信息生成结构清晰、重点突出、无冗余信息的每日项目进度报告。 工作原则 1. 优先使用我提供的工具来获取最新、最准确的数据。 2. 对获取的数据进行交叉验证和简要分析例如提交数量是否异常是否有高风险问题卡住。 3. 报告格式请严格使用Markdown包含以下章节概览、代码活动、任务状态、风险与阻塞、今日建议。 4. 语言风格保持专业、简洁、积极。对于发现的问题客观描述并尝试提供建设性意见。 5. 只有当我明确要求或者报告最终版已生成且确认无误后才使用send_slack_report工具发送报告。 如果任何工具调用失败或返回空数据请在报告中明确说明“XXX数据暂不可用”并基于已有信息继续完成报告。这个提示词明确了角色、任务、步骤、格式和风格极大地缩小了模型“胡思乱想”的空间。6.3 成本与性能优化Claude API是按Token收费的智能体对话由于包含工具调用和长上下文Token消耗可能不小。控制上下文长度定期清理conversation_history。对于定时报告这种独立任务每次运行都可以使用全新的对话历史避免携带无关旧信息。压缩技能返回数据如前所述只返回必要信息。不要将完整的API响应可能包含大量无用字段直接扔给模型。设定清晰的终止条件在用户查询或系统提示中明确任务边界避免模型陷入无限追问或过度展开的循环。例如在请求生成报告后模型不应再反问“您还需要我做什么”。使用更合适的模型对于主要依赖工具调用、逻辑相对固定的任务可以尝试使用claude-3-haiku模型。它速度更快成本更低在遵循清晰指令执行标准化任务时表现不错。可以在开发调试阶段用Sonnet生产环境考虑Haiku。6.4 一个真实的踩坑案例时区问题我的Bot在本地测试时报告显示“昨日提交”一切正常。但部署到位于UTC时区的云服务器后每天上午10点CST运行时since_time计算的是UTC时间的“昨天”这导致它漏掉了UTC时间当天凌晨即CST时间昨天下午的提交。排查首先检查日志发现GitHub技能返回的数据为空。然后打印出计算的since_time参数发现是UTC时间。与本地时间对比立刻发现了时区差异。解决在技能处理函数中显式指定时区。from datetime import datetime, timedelta, timezone # 使用东八区时间 tz_shanghai timezone(timedelta(hours8)) since_time (datetime.now(tz_shanghai) - timedelta(days1)).isoformat()或者更通用的做法是从环境变量读取应用运行的时区配置。经验所有与时间相关的逻辑数据查询、报告生成时间戳在开发初期就必须考虑时区问题并统一使用UTC时间或一个明确的时区进行存储和计算避免因部署环境变化导致的数据错乱。走到这一步一个能够自动收集信息、分析数据、生成并发送日报的Bot智能体就已经初具雏形了。它不再是一个简单的脚本而是一个能够理解复杂指令、自主协调多个外部工具、并做出一定分析的“准智能”助手。你可以在此基础上继续为它添加更多技能比如从Confluence读取项目文档、从监控系统拉取性能指标甚至让它根据报告内容自动创建下周的规划任务。智能体开发的魅力就在于它的能力边界最终只取决于你为它连接了多少个“手”和“眼睛”。 基于Claude Code Skills构建AI智能体：从原理到自动化项目进度Bot实战
1. 从“玩具”到“生产力”为什么现在必须掌握智能体开发如果你最近在技术社区里泡着大概率会频繁刷到“AI智能体”这个词。从年初的AutoGPT、BabyAGI到后来各大模型厂商推出的“GPTs”、“Copilot Studio”再到如今Claude、GPT-4o等模型原生支持的“Code Skills”或“Function Calling”能力智能体已经从实验室里的概念变成了触手可及的生产力工具。很多人可能觉得这不就是让AI调用几个API吗有什么大不了的我最初也是这么想的直到我亲手用Claude Code Skills在几个小时内把一个需要反复手动查询、复制粘贴数据的日报生成流程自动化成了一个能自主运行、定时汇报的Bot我才意识到这件事的本质变了。过去我们让AI写代码、回答问题它更像一个“超级实习生”你给指令它产出内容但执行和串联的“手”和“脑”还是你自己。而智能体是让AI自己拥有了“手”和“眼睛”。这个“手”就是Code Skills——模型直接编写并执行代码的能力。这意味着AI不再只是给你一段需要你手动去跑的Python脚本而是可以自己思考“要完成用户‘获取今日销售额并分析趋势’这个任务我需要先调用公司CRM的API技能A拿到数据后再用matplotlib画个图技能B最后把图表和总结发到Slack频道技能C。” 然后它自己就把这一套流程给执行了。所以“从零开发一个Bot智能体”在今天已经不是一个炫技的玩具项目而是一个实实在在的效能倍增器。它解决的核心痛点是将复杂、多步骤、跨工具的手动操作封装成一个由自然语言驱动的、可重复执行的自动化流程。无论是每天早上的数据拉取与简报、监控告警的自动分析与初步排查还是跨系统信息的查询与整合一个设计良好的Bot智能体都能让你从重复劳动中彻底解放出来。Claude的Code Skills在这一波浪潮中显得尤为突出因为它背后是Claude 3系列模型强大的代码生成与推理能力。与简单的函数调用不同Code Skills允许模型在一个持久的、安全的沙箱环境中动态生成并运行代码处理文件安装临时依赖。这相当于给了AI一个功能完整的“临时工作台”其灵活性和解决问题的能力上了一个大台阶。接下来我就以打造一个“项目进度追踪与自动汇报Bot”为例带你完整走一遍实战流程分享从环境准备、技能设计、到调试部署的每一个细节以及我趟过的那些坑。2. 战场准备理解Claude Code Skills的运行机制与边界在撸起袖子写第一个技能之前我们必须先摸清楚手中的武器——Claude Code Skills——到底是怎么工作的以及它的能力边界在哪里。这决定了我们设计智能体的思路避免后期出现“我以为它能结果它不能”的尴尬局面。2.1 核心机制沙箱、会话与技能库Claude Code Skills不是让AI模型在你的本地终端或服务器上为所欲为。它的核心是一个受控的代码执行沙箱环境。当你通过API或Claude Console触发一个Code Skills调用时会发生以下几步意图识别Claude模型首先理解你的自然语言请求例如“帮我看看上周项目日志里ERROR级别的记录有多少条”。技能匹配与规划模型会判断是否需要以及如何使用已配置的技能。它会思考“用户需要分析日志文件。我有一个‘文件读取’技能可以读取指定路径的文件内容。我还需要一个‘文本分析’技能来统计ERROR关键词。我需要先调用技能A将其结果作为输入传递给技能B。”代码生成与执行对于需要Code Skills的任务模型会在沙箱中动态生成Python代码这是目前的主要语言。这个沙箱是临时的、隔离的通常具备基础Python环境、网络访问能力可配置和有限的临时磁盘空间。代码在此沙箱中运行。结果返回代码执行的标准输出stdout、错误信息stderr以及可能的文件产出会作为结果返回给Claude模型。模型再对这些结果进行解读、总结并用自然语言回复给你。这里的关键在于“技能”Skills的配置。你可以预先定义好一系列技能每个技能本质上是一个允许模型执行的操作的抽象描述。例如一个“HTTP请求”技能告诉模型“你可以向互联网发送GET/POST请求来获取数据”。一个“数据库查询”技能告诉模型“你可以连接到一个安全的数据库通过预配置的连接信息执行SQL查询”。一个“运行Shell命令”技能告诉模型“你可以在沙箱中执行特定的、允许的系统命令”。模型根据这些技能描述来决定如何组合它们以完成任务。Code Skills则是更底层、更灵活的一环当预定义的技能不够用时模型可以直接编写代码来实现复杂逻辑。2.2 能力边界与安全红线理解边界比理解能力更重要这直接关系到项目的可行性与安全性。持久化与状态每次Code Skills调用沙箱环境通常是全新的。这意味着你不能指望在一次会话中在/tmp目录下写一个文件然后在下次调用中还能读到它。智能体的“状态”需要通过外部存储来维护比如数据库、文件存储服务S3等或你在技能中提供的API。执行时间与资源沙箱有执行时间限制例如30秒或60秒和内存/CPU限制。长时间运行的后台任务、需要大量计算的任务如训练机器学习模型不适合在此运行。网络访问沙箱可以访问外网但这通常需要显式启用或配置。出于安全考虑某些内部网络地址或端口可能会被阻止。文件系统访问沙箱对文件系统的访问是受限的通常只能访问临时目录或你通过技能特别挂载的路径。它不能随意读写宿主机上的任意文件。安全性最重要这是双刃剑。沙箱保护了你的主机环境但你也必须警惕提示词注入避免让用户输入直接成为代码或命令的一部分。永远要对用户提供的参数进行严格的校验和清理。敏感信息泄露绝对不要将API密钥、数据库密码等硬编码在技能描述或可能被模型生成的代码中。应该使用环境变量或安全的密钥管理服务并通过技能配置以安全的方式传递给沙箱环境。无限循环与资源耗尽模型生成的代码可能有bug比如死循环。沙箱通常有看门狗机制但设计技能时也要有意识避免此类模式。提示在开发初期一个非常实用的做法是在技能描述中明确写出约束条件。例如在“执行数据分析”的技能描述里加上“注意沙箱环境最多使用1GB内存运行时间不超过30秒。请确保生成的Pandas代码不会尝试读取超过100MB的文件。”2.3 我们的Bot智能体蓝图基于以上理解我们来规划要打造的“项目进度自动汇报Bot”的核心功能与架构核心目标每天上午10点自动收集指定项目的关键数据如Git提交记录、JIRA问题状态、CI/CD流水线结果生成一份结构化的进度报告并发送到团队Slack频道。架构设计触发器一个定时任务例如使用云函数、cron job或Make.com/Zapier等自动化平台定期调用我们的Bot智能体。智能体核心一个封装了Claude Code Skills的程序。它接收触发由Claude模型理解任务并协调执行一系列技能。技能集技能AGit仓库查询。通过GitHub/GitLab API获取昨日提交列表、分支合并情况。技能B项目管理JIRA数据获取。查询指定JIRA看板获取“进行中”、“待测试”、“已完成”状态的任务列表。技能CCI/CD状态检查。调用Jenkins/GitLab CI API获取最新构建的成功/失败状态。技能D数据分析与报告生成。将A、B、C技能获取的原始数据进行聚合、分析如计算完成度、阻塞项并格式化为Markdown或HTML报告。技能E消息推送。将生成的报告发送到Slack指定频道。状态/缓存层一个简单的数据库如SQLite或键值存储如Redis用于记录上次运行的时间、已处理的数据ID避免报告重复内容。这个架构将利用Code Skills的灵活性来处理数据分析和报告生成技能D而其他相对固定的API调用则可以用预定义的HTTP技能或封装好的函数来实现。3. 搭建开发环境从API Key到第一个“Hello World”技能理论说再多不如动手跑通第一个例子。这里我选择使用Anthropic官方提供的Python SDK进行开发因为它最直接也便于我们理解底层交互。3.1 获取必要的密钥与工具Anthropic API Key前往 Anthropic Console 注册并创建API Key。这是调用Claude模型的通行证。妥善保管不要提交到代码仓库。Python环境建议使用Python 3.9。使用venv或conda创建一个干净的虚拟环境。python -m venv claude-bot-env source claude-bot-env/bin/activate # Linux/macOS # claude-bot-env\Scripts\activate # Windows安装SDK在虚拟环境中安装Anthropic Python SDK。pip install anthropic3.2 编写第一个“代码执行”对话我们先不急着定义复杂的技能而是直接让Claude在沙箱里运行一段代码感受一下Code Skills最原始的能力。创建一个文件first_skill.pyimport anthropic import os # 从环境变量读取API Key更安全 client anthropic.Anthropic( api_keyos.environ.get(ANTHROPIC_API_KEY) ) # 构建一个要求Claude执行代码的对话 message client.messages.create( modelclaude-3-5-sonnet-20241022, # 使用支持Code Skills的最新模型 max_tokens1000, system你是一个有帮助的AI助手可以使用Code Skills执行Python代码来解决问题。, messages[ { role: user, content: 请帮我计算一下斐波那契数列的前10个数字并用Python代码实现告诉我结果。 } ] ) print(message.content[0].text)运行这个脚本记得先设置环境变量ANTHROPIC_API_KEY你会看到Claude不仅给出了斐波那契数列的代码还输出了执行结果[0, 1, 1, 2, 3, 5, 8, 13, 21, 34]。这说明模型在背后已经生成了代码并在沙箱中执行完毕。但这还不够“技能化”。我们更希望模型能调用我们预先定义好的、更安全可控的功能。3.3 定义并集成第一个自定义技能文件内容读取让我们定义一个更实用的技能读取指定项目目录下的CHANGELOG.md文件并总结最新版本更新。我们将通过SDK的tools参数来定义这个技能。首先我们按照Anthropic的tools格式定义一个技能。这本质上是一个符合OpenAI Function Calling格式的JSON Schema描述。import anthropic import os client anthropic.Anthropic(api_keyos.environ.get(ANTHROPIC_API_KEY)) # 1. 定义“读取文件”技能 read_file_tool { name: read_project_changelog, description: 读取项目根目录下的CHANGELOG.md文件内容用于分析最近的版本更新。, input_schema: { type: object, properties: {} # 这个技能不需要输入参数固定读取一个文件 required: [] } } # 2. 在实际对话中当模型决定调用此技能时我们需要一个处理函数来执行真正的逻辑 def handle_read_changelog(): 模拟读取CHANGELOG文件的处理函数 # 这里应该是真实的文件读取逻辑。例如 # with open(/path/to/project/CHANGELOG.md, r) as f: # content f.read() # 为了演示我们返回模拟数据 mock_changelog # 项目变更日志 ## [v1.2.0] - 2024-04-15 ### 新增 - 实现了用户权限管理模块。 - 添加了项目导出为PDF功能。 ## [v1.1.0] - 2024-04-01 ### 修复 - 解决了登录页面在移动端的布局错乱问题。 - 修复了数据导出的时间戳错误。 return {content: mock_changelog} # 3. 发起对话并提供技能定义 message client.messages.create( modelclaude-3-5-sonnet-20241022, max_tokens1000, system你是一个项目助理Bot。你可以使用工具来获取项目信息。, tools[read_file_tool], # 关键在这里传入技能定义 messages[ { role: user, content: 请告诉我这个项目最近一个版本更新了什么内容 } ] ) # 4. 检查模型的回复 initial_response message.content[0].text print(Claude的初始回复可能要求调用技能: , initial_response) # 在实际的流式或异步交互中模型可能会返回一个tool_use块表示它想调用某个技能。 # 我们需要检查响应如果模型要求调用工具则执行对应的处理函数并将结果追加到对话中再次请求模型。 # 下面是一个简化的同步示例逻辑 if hasattr(message, tool_calls) and message.tool_calls: # 注意实际API响应结构可能不同此处为逻辑示意 # 假设模型决定调用 read_project_changelog tool_result handle_read_changelog() # 然后将工具执行结果作为新的消息内容再次发送给模型进行总结 second_message client.messages.create( modelclaude-3-5-sonnet-20241022, max_tokens1000, system你是一个项目助理Bot。, tools[read_file_tool], messages[ {role: user, content: 请告诉我这个项目最近一个版本更新了什么内容}, {role: assistant, content: f我需要调用read_project_changelog技能来获取信息。}, # 模拟助理的请求 { role: user, # 注意工具执行结果通常以tool角色发送 content: [{ type: tool_result, tool_use_id: message.tool_calls[0].id, # 需要匹配tool call的ID content: tool_result[content] }] } ] ) print(\nClaude在获取文件内容后的总结\n, second_message.content[0].text) else: print(\nClaude直接给出了回答\n, initial_response)注意上述代码中关于tool_calls的处理是一个逻辑示意。Anthropic Messages API对于工具调用的实际处理是流式streaming的或者需要在收到tool_use区块后手动构造后续请求。最新的SDK和API文档提供了更优雅的tools集成方式。这里的目的是展示“定义技能”和“模型决定使用技能”的基本交互模式。在实际开发中你需要根据官方指南实现完整的工具调用循环。通过这一步我们明确了技能开发的基本模式定义技能描述 - 模型在对话中请求调用 - 我们的后端执行实际逻辑 - 将结果返回给模型 - 模型生成最终回复。接下来我们就要用这个模式构建我们Bot的核心技能集。4. 构建核心技能集让Bot拥有“手”和“眼睛”我们的项目进度Bot需要与多个外部系统交互。为每个系统封装一个清晰的技能是智能体稳定可靠的基础。这里我以GitHub API和Slack Incoming Webhook为例展示两个最关键技能的完整实现。4.1 技能一GitHub提交记录获取器这个技能的目标是根据给定的仓库名、日期范围获取相关的提交记录。首先我们需要一个安全的密钥管理方式。我们将使用python-dotenv来管理环境变量。安装依赖并配置环境pip install python-dotenv requests在项目根目录创建.env文件ANTHROPIC_API_KEYsk-ant-... GITHUB_TOKENghp_... # 需要repo权限的GitHub Personal Access Token GITHUB_OWNERyour_company_or_username.env文件务必加入.gitignore避免密钥泄露。实现GitHub API交互函数 创建一个模块github_integration.pyimport os import requests from datetime import datetime, timedelta from dotenv import load_dotenv load_dotenv() GITHUB_API https://api.github.com HEADERS { Authorization: ftoken {os.getenv(GITHUB_TOKEN)}, Accept: application/vnd.github.v3json } def get_recent_commits(repo_name: str, since_days: int 1) - list: 获取指定仓库最近N天内的提交记录。 Args: repo_name: 仓库名称不含所有者。 since_days: 查询多少天内的数据默认1天昨日至今。 Returns: 一个包含提交信息的字典列表。 since_time (datetime.now() - timedelta(dayssince_days)).isoformat() url f{GITHUB_API}/repos/{os.getenv(GITHUB_OWNER)}/{repo_name}/commits params {since: since_time, per_page: 50} # 限制每页数量 try: response requests.get(url, headersHEADERS, paramsparams) response.raise_for_status() # 如果状态码不是200抛出异常 commits response.json() # 简化返回数据只提取关键信息 simplified_commits [] for commit in commits: commit_info commit.get(commit, {}) author commit_info.get(author, {}) simplified_commits.append({ sha: commit.get(sha, )[:7], message: commit_info.get(message, ).split(\n)[0], # 取第一行作为概要 author: author.get(name, N/A), date: author.get(date, ), url: commit.get(html_url, ) }) return simplified_commits except requests.exceptions.RequestException as e: print(fError fetching commits from GitHub: {e}) return [] except ValueError as e: print(fError parsing GitHub response: {e}) return []将函数封装为Claude可调用的技能 在定义给Claude的tools列表中我们需要用JSON Schema来描述这个技能。# 在 main_bot.py 或类似主逻辑文件中 github_tool { name: get_github_commits, description: 获取指定GitHub仓库在最近一段时间内的提交记录。用于追踪代码变更活动。, input_schema: { type: object, properties: { repo_name: { type: string, description: GitHub仓库的名称例如my-web-app }, since_days: { type: integer, description: 查询从多少天前到现在默认是1即昨天到今天, default: 1 } }, required: [repo_name] # repo_name是必填参数 } } # 对应的处理函数 def handle_get_github_commits(repo_name: str, since_days: int 1): from github_integration import get_recent_commits # 实际项目中注意导入 commits get_recent_commits(repo_name, since_days) # 返回给Claude的结果需要是字符串或可序列化结构 if commits: # 可以格式化一下方便模型阅读 result_lines [f- {c[sha]}: {c[message]} by {c[author]} for c in commits[:10]] # 只展示前10条 return {commits: commits, summary: f共找到{len(commits)}条提交。最近几条\n \n.join(result_lines)} else: return {commits: [], summary: 在指定时间内未找到提交记录或查询出错。}现在当Claude模型认为需要获取GitHub提交信息时它会生成一个类似{repo_name: my-web-app, since_days: 1}的调用请求。我们的后端程序捕获到这个请求执行handle_get_github_commits函数并将结果返回给模型。4.2 技能二Slack消息推送器Bot生成报告后需要能推送到指定频道。我们使用Slack的Incoming Webhooks它比Bot Token更简单适合单向发送消息。在Slack创建Incoming Webhook访问 api.slack.com/apps 。创建新App或选择现有App在功能列表中找到“Incoming Webhooks”并激活。添加一个新的Webhook到你的目标频道复制生成的Webhook URL。将Webhook URL加入环境变量 在.env文件中添加SLACK_WEBHOOK_URLhttps://hooks.slack.com/services/...实现Slack推送函数 创建slack_integration.pyimport os import json import requests from dotenv import load_dotenv load_dotenv() def send_slack_message(text: str, blocks: list None, channel: str None): 通过Incoming Webhook发送消息到Slack。 Args: text: 消息的fallback文本。 blocks: Slack Block Kit格式的消息块用于富文本格式。 channel: 可覆盖默认频道的目标频道如#general。需Webhook已授权。 webhook_url os.getenv(SLACK_WEBHOOK_URL) if not webhook_url: print(错误未配置SLACK_WEBHOOK_URL环境变量。) return False payload {text: text} if blocks: payload[blocks] blocks if channel: payload[channel] channel headers {Content-Type: application/json} try: response requests.post(webhook_url, datajson.dumps(payload), headersheaders) response.raise_for_status() print(f消息已成功发送至Slack。) return True except requests.exceptions.RequestException as e: print(f发送消息到Slack失败: {e}) return False封装为Claude技能slack_tool { name: send_slack_report, description: 将格式化好的项目进度报告发送到指定的Slack频道。, input_schema: { type: object, properties: { report_markdown: { type: string, description: 完整的项目进度报告使用Markdown格式。 }, channel_override: { type: string, description: 可选指定发送到哪个Slack频道例如#project-updates。不填则使用Webhook默认频道。 } }, required: [report_markdown] } } def handle_send_slack_report(report_markdown: str, channel_override: str None): from slack_integration import send_slack_message # 将Markdown转换为Slack的mrkdwn格式Slack基本支持Markdown # 也可以使用更复杂的Block Kit来美化消息 blocks [ { type: section, text: { type: mrkdwn, text: f* 项目每日进度报告*\n{report_markdown} } } ] success send_slack_message( textf项目每日进度报告{report_markdown[:100]}..., # fallback文本 blocksblocks, channelchannel_override ) return {success: success, message: 报告已尝试发送至Slack。}按照同样的模式我们可以继续添加JIRA查询、CI状态检查等技能。关键在于每个技能的描述description要清晰准确输入参数input_schema要定义明确。这相当于给Claude模型一本清晰的“工具说明书”它才能正确地决定在什么情况下使用哪把“扳手”。5. 组装与调度构建智能体的“大脑”与工作流有了一个个独立的技能我们现在需要创建一个“大脑”——也就是主控程序来协调Claude模型和这些技能并设计一个触发它工作的机制。5.1 实现智能体主循环智能体的核心是一个循环接收用户输入或定时触发- 调用Claude模型并告知可用技能 - 解析模型响应 - 若模型要求调用技能则执行并返回结果 - 将结果反馈给模型继续对话直到模型给出最终答案。下面是一个简化但完整的主循环示例bot_agent.pyimport anthropic import os import json from dotenv import load_dotenv from typing import Dict, Any, List # 导入技能处理函数 from my_skills.github_skill import handle_get_github_commits from my_skills.slack_skill import handle_send_slack_report # ... 导入其他技能处理函数 load_dotenv() class ProjectReportBot: def __init__(self): self.client anthropic.Anthropic(api_keyos.getenv(ANTHROPIC_API_KEY)) # 定义所有可用的技能工具 self.tools [ { # GitHub技能定义 name: get_github_commits, description: 获取指定GitHub仓库在最近一段时间内的提交记录。用于追踪代码变更活动。, input_schema: { type: object, properties: { repo_name: {type: string, description: GitHub仓库名称}, since_days: {type: integer, description: 查询天数, default: 1} }, required: [repo_name] } }, { # Slack技能定义 name: send_slack_report, description: 将格式化好的项目进度报告发送到指定的Slack频道。, input_schema: { type: object, properties: { report_markdown: {type: string, description: Markdown格式报告}, channel_override: {type: string, description: 可选频道} }, required: [report_markdown] } } # ... 添加其他工具定义 ] # 映射工具名到处理函数 self.tool_handlers { get_github_commits: handle_get_github_commits, send_slack_report: handle_send_slack_report, # ... 映射其他工具 } self.conversation_history [] # 用于维护对话上下文 def _execute_tool(self, tool_name: str, tool_input: Dict[str, Any]) - Dict[str, Any]: 执行具体的工具/技能 handler self.tool_handlers.get(tool_name) if not handler: return {error: f未知的工具: {tool_name}} try: # 将字典参数解包传递给处理函数 return handler(**tool_input) except Exception as e: return {error: f执行工具{tool_name}时出错: {str(e)}} def run(self, user_query: str) - str: 运行智能体处理一次用户查询 # 1. 将用户查询加入历史 self.conversation_history.append({role: user, content: user_query}) # 2. 准备发送给Claude的消息包含全部历史 message_payload { model: claude-3-5-sonnet-20241022, max_tokens: 4000, system: 你是一个高效的项目进度助理Bot。你的任务是整合来自GitHub、JIRA等多个来源的信息生成清晰、简洁的每日项目进度报告并可以应要求将其发送到Slack。请充分利用我提供给你的工具。, tools: self.tools, messages: self.conversation_history } # 3. 发送请求并处理流式响应以处理工具调用 with self.client.messages.stream(**message_payload) as stream: final_text for event in stream: # 处理文本流 if event.type content_block_delta and event.delta.type text_delta: final_text event.delta.text # 处理工具调用请求这是关键部分 elif event.type content_block_start and event.content_block.type tool_use: tool_use_block event.content_block tool_name tool_use_block.name tool_input tool_use_block.input print(f[Bot] Claude请求调用工具: {tool_name} 参数: {tool_input}) # 4. 执行工具 tool_result self._execute_tool(tool_name, tool_input) # 将结果转换为字符串格式方便Claude读取 result_content json.dumps(tool_result, ensure_asciiFalse, indent2) print(f[Bot] 工具执行结果: {result_content[:200]}...) # 5. 将工具执行结果作为新的消息块发送回去 stream.send({ type: tool_result, tool_use_id: tool_use_block.id, content: result_content }) # 6. 将Claude的最终回复加入历史维持上下文可选取决于是否需要多轮对话 if final_text: self.conversation_history.append({role: assistant, content: final_text}) return final_text # 使用示例 if __name__ __main__: bot ProjectReportBot() # 模拟一个复杂的用户请求 query 请生成一份关于my-web-app仓库的昨日项目进度报告。 报告需要包含 1. 昨天所有的代码提交概要。 2. 如果提交数量大于5请分析一下开发活跃度。 3. 最后将这份报告用Markdown格式整理好并发送到Slack的#daily-standup频道。 report bot.run(query) print(\n *50) print(最终生成的报告) print(report)这个run方法利用Anthropic SDK的流式接口能够实时处理模型中途发起的工具调用请求形成一个完整的“思考-行动-观察”循环。5.2 设计定时触发与部署我们的Bot需要自动运行而不是手动执行脚本。这里有几个经典的部署方案方案A云函数/Serverless推荐平台Vercel Serverless Functions AWS Lambda Google Cloud Functions。流程将上述ProjectReportBot类打包。创建一个HTTP端点如/api/generate-daily-report。使用云服务商的定时触发器如AWS EventBridge Google Cloud Scheduler Vercel Cron Jobs每天定点调用这个端点。优点无需管理服务器按需付费伸缩性好。注意需要将环境变量API Keys安全地配置在云平台中。方案B传统服务器 Cron Job流程在一台长期运行的服务器或容器上部署Python脚本。使用Linux的crontab设置定时任务例如每天上午10点运行0 10 * * * /path/to/venv/bin/python /path/to/bot_agent.py /var/log/bot.log 21。优点控制力强调试方便。缺点需要维护服务器。方案C低代码自动化平台平台Make.com, Zapier, n8n。流程在这些平台上设置一个定时触发器触发一个HTTP请求Webhook到你的Bot服务端点可以是方案A的云函数也可以是一个简单的服务器。或者如果逻辑不复杂可以直接在这些平台上用内置模块调用Claude API并处理返回结果。优点配置可视化适合非开发人员或快速原型。我个人的选择是方案AVercel Serverless Cron Job。将核心逻辑部署为Serverless Function通过Vercel提供的Cron Jobs来触发几乎零运维成本极低。你需要做的就是把代码推送到GitHub并连接Vercel项目。6. 避坑指南与效能优化来自实战的经验之谈在开发和运行这个Bot的过程中我踩过不少坑也总结出一些让智能体更可靠、更高效的经验。6.1 技能设计中的三个常见陷阱技能描述过于模糊或宽泛反面例子处理数据。这种描述让模型无法准确判断何时使用。正面例子查询指定GitHub仓库在特定时间范围内的提交记录并返回提交哈希、作者、信息和时间。清晰描述了功能、输入和输出。技巧在描述中明确技能的目的、输入格式和输出格式。甚至可以加入使用示例或约束条件例如“此技能用于获取昨日至今的提交输入参数repo_name必须是字符串返回一个列表。”技能返回数据格式混乱问题处理函数返回一个复杂的Python对象如包含datetime对象的字典直接json.dumps可能会失败。或者返回的字符串过于冗长干扰模型的总结。解决在返回给模型前对数据进行序列化和精简。确保所有数据都是JSON可序列化的字符串、数字、列表、字典。提取最关键的信息过滤掉无关的元数据。例如GitHub API返回的提交信息很多我们只提取了sha、message、author、date这几个字段。缺乏错误处理与降级方案场景GitHub API临时不可用或者JIRA查询超时。如果技能直接抛出异常整个Bot流程就会中断。解决在每个技能处理函数内部做好健壮的错误处理try-catch。即使出错也返回一个结构化的错误信息而不是让程序崩溃。例如return {error: GitHub API请求失败状态码XXX, data: []}。这样模型还能在最终报告里提及“GitHub数据暂时无法获取”而不是完全卡住。6.2 提示词Prompt工程引导模型做出最佳决策系统提示词system是智能体的“人格”和“工作指南”写得好坏直接影响输出质量。基础版不够好“你是一个助手。”优化版针对我们的Bot你是一个专注、高效、严谨的项目进度助理Bot。你的核心任务是整合来自多个源头GitHub, JIRA的信息生成结构清晰、重点突出、无冗余信息的每日项目进度报告。 工作原则 1. 优先使用我提供的工具来获取最新、最准确的数据。 2. 对获取的数据进行交叉验证和简要分析例如提交数量是否异常是否有高风险问题卡住。 3. 报告格式请严格使用Markdown包含以下章节概览、代码活动、任务状态、风险与阻塞、今日建议。 4. 语言风格保持专业、简洁、积极。对于发现的问题客观描述并尝试提供建设性意见。 5. 只有当我明确要求或者报告最终版已生成且确认无误后才使用send_slack_report工具发送报告。 如果任何工具调用失败或返回空数据请在报告中明确说明“XXX数据暂不可用”并基于已有信息继续完成报告。这个提示词明确了角色、任务、步骤、格式和风格极大地缩小了模型“胡思乱想”的空间。6.3 成本与性能优化Claude API是按Token收费的智能体对话由于包含工具调用和长上下文Token消耗可能不小。控制上下文长度定期清理conversation_history。对于定时报告这种独立任务每次运行都可以使用全新的对话历史避免携带无关旧信息。压缩技能返回数据如前所述只返回必要信息。不要将完整的API响应可能包含大量无用字段直接扔给模型。设定清晰的终止条件在用户查询或系统提示中明确任务边界避免模型陷入无限追问或过度展开的循环。例如在请求生成报告后模型不应再反问“您还需要我做什么”。使用更合适的模型对于主要依赖工具调用、逻辑相对固定的任务可以尝试使用claude-3-haiku模型。它速度更快成本更低在遵循清晰指令执行标准化任务时表现不错。可以在开发调试阶段用Sonnet生产环境考虑Haiku。6.4 一个真实的踩坑案例时区问题我的Bot在本地测试时报告显示“昨日提交”一切正常。但部署到位于UTC时区的云服务器后每天上午10点CST运行时since_time计算的是UTC时间的“昨天”这导致它漏掉了UTC时间当天凌晨即CST时间昨天下午的提交。排查首先检查日志发现GitHub技能返回的数据为空。然后打印出计算的since_time参数发现是UTC时间。与本地时间对比立刻发现了时区差异。解决在技能处理函数中显式指定时区。from datetime import datetime, timedelta, timezone # 使用东八区时间 tz_shanghai timezone(timedelta(hours8)) since_time (datetime.now(tz_shanghai) - timedelta(days1)).isoformat()或者更通用的做法是从环境变量读取应用运行的时区配置。经验所有与时间相关的逻辑数据查询、报告生成时间戳在开发初期就必须考虑时区问题并统一使用UTC时间或一个明确的时区进行存储和计算避免因部署环境变化导致的数据错乱。走到这一步一个能够自动收集信息、分析数据、生成并发送日报的Bot智能体就已经初具雏形了。它不再是一个简单的脚本而是一个能够理解复杂指令、自主协调多个外部工具、并做出一定分析的“准智能”助手。你可以在此基础上继续为它添加更多技能比如从Confluence读取项目文档、从监控系统拉取性能指标甚至让它根据报告内容自动创建下周的规划任务。智能体开发的魅力就在于它的能力边界最终只取决于你为它连接了多少个“手”和“眼睛”。