2026年带你实现一个Agent,从Tools、MCP到Skills,从理论到代码,干货满满

带你实现一个Agent,从Tools、MCP到Skills,从理论到代码,干货满满盘点 25 年最火的两个事件 我会选择 DeepSeek 发布和 Manus 发布 市场今年是一看到这类产品就很兴奋 包括最近爆火的 Agent 变形产品 OpenClaw Clawdbot Moltbot 至于为什么需要 Agent 他到底解决了什么问题 我们在之前的文章里面有详细介绍 今天的话 我们更加务实一些 直接手把手的教大家如何做一个 Agent 由此会让大家更加理解 Agent 的难点到底在哪

大家好,我是讯享网,很高兴认识大家。这里提供最前沿的Ai技术和互联网信息。



盘点25年最火的两个事件,我会选择DeepSeek发布和Manus发布,市场今年是一看到这类产品就很兴奋,包括最近爆火的Agent变形产品:OpenClaw(Clawdbot→Moltbot)。

至于为什么需要Agent,他到底解决了什么问题,我们在之前的文章里面有详细介绍:

今天的话,我们更加务实一些,直接手把手的教大家如何做一个Agent,由此会让大家更加理解Agent的难点到底在哪:

在开发中,不同项目需要不同版本的 Python(例如 3.10、3.12)。为了方便管理多版本 Python,我们使用 pyenv。

官方地址:https://github.com/pyenv/pyenv

macOS 使用 Homebrew 安装:

brew update brew install pyenv 

或者通过脚本安装:

curl https://pyenv.run | bash 

然后就是Linux和Windows 安装了:

curl -fsSL https://pyenv.run | bash

Windows 使用 pyenv-win:

  1. 打开 PowerShell(管理员权限)。
  2. 执行安装命令: Invoke-WebRequest -UseBasicParsing -Uri "https://raw.githubusercontent.com/pyenv-win/pyenv-win/master/pyenv-win/install-pyenv-win.ps1" -OutFile "./install-pyenv-win.ps1"; &"./install-pyenv-win.ps1"

    安装完成后,运行以下命令查看版本:

    pyenv –version 

    如果显示版本号,则安装成功。以下是一些常见命令:

    # 查看可安装的Python版本 pyenv install –list | grep 3.12

    安装Python 3.12.0

    pyenv install 3.12.0

    查看已安装的版本

    pyenv versions

    设置全局Python版本

    pyenv global 3.12.0

    为当前目录设置Python版本

    pyenv local 3.12.0

    Python 包管理工具

    uv 是一个轻量级的 Python 包管理工具,用于管理虚拟环境和依赖包,类似 pip + venv 的组合。使用 uv 可以方便地为不同项目创建独立的虚拟环境,并管理依赖版本。

    官方地址:https://github.com/astral-sh/uv

    确保已经安装好 Python(推荐通过 pyenv 管理不同版本):

    pyenv install 3.12.0 pyenv global 3.12.0 # 或者在项目目录使用 pyenv local 3.12.0 

    macOS / Linux 通过官方脚本安装,或者使用 pip 安装::

    curl -fsSL https://uv.dev/install.sh | bash pip install uv-cli 

    多数同学可能是Windows,这里安装有两个步骤:

    pip install uv-cli 

    安装完成后,运行以下命令检查版本,如果显示版本号,则说明安装成功:

    uv –version 

    常用命令如下:

    # 初始化项目(自动创建虚拟环境) uv init

    查看当前项目使用的 Python 版本

    uv run python –version

    安装依赖包

    uv add requests flask uv add openai

    卸载依赖包

    uv remove requests

    锁定当前环境依赖(生成 uv.lock)

    uv lock

    根据锁文件安装依赖

    uv sync

    pyCharm开发工具

    最后是开发工具的安装,官网下载地址 https://www.jetbrains.com/pycharm/download/

    首先我们后面会用到很多第三方服务,这里需要一些Key:

    一、访问 DeepSeek开发者平台 注册登录,创建一个API Key用于大模型调用:

    二、访问高德开放平台官网创建应用并选择Web服务类型,获取的Key将用于查询酒店、景点、路径规划和天气信息,个人开发者有一定的免费调用额度,完全满足日常使用需求:

    三、将获取到的key配置到.env 文件中,或者配置到系统环境变量中:

    echo "DEEPSEEK_API_KEY=youkey" > .env echo "AMAP_API_KEY=youkey" >> .env # 注意这里是两个大于号 

    这里可以简单测试下:

    # deepseek def deepseekDemo(): print("deepseekDemo,hello") 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)

高德

def amap_demo(): key = os.getenv("AMAP_API_KEY") response = httpx.Client().get(

 "https://restapi.amap.com/v3/weather/weatherInfo?key=" + key + "&city=&extensions=all") 

print(response.json()) print(json.dumps(response.json(), indent=2,ensure_ascii=False))

基础环境OK了,我们就来看Agent的能力基石Function Calling:

大模型本身只能做文本理解和生成,无法直接访问数据或执行外部逻辑,例如查询天气、搜索景点、计算路线等。

Tools(函数调用)机制的作用,是由应用侧提供一组可调用的函数,在模型推理过程中,由模型决定是否需要调用这些函数、以及调用哪一个、使用什么参数。

模型只负责决策,应用程序负责真正执行函数并返回结果:

Tools 的调用流程

Tools 的调用并不是一次完成的,而是一个多轮交互过程:

1.发起第一次模型调用

应用程序首先向大模型发起一个包含用户问题与模型可调用工具清单的请求。

2.接收模型的工具调用指令(工具名称与入参)

若模型判断需要调用外部工具,会返回一个JSON格式的指令,用于告知应用程序需要执行的函数与入参;

若模型判断无需调用工具,会返回自然语言格式的回复。

3.在应用端运行工具

应用程序接收到工具指令后,需要运行工具,获得工具输出结果。

4.发起第二次模型调用

获取到工具输出结果后,需添加至模型的上下文(messages),再次发起模型调用。

5.接收来自模型的最终响应

模型将工具输出结果和用户问题信息整合,生成自然语言格式的回复。

为什么要使用 Tools

在早期没有标准化工具调用能力时,Agent主要依赖提示词驱动。

开发者需要提前实现所有功能函数,再通过复杂提示词告诉模型有哪些函数、什么时候用、参数怎么传。

模型通常以自然语言或半结构化文本的形式输出“行动计划”,应用程序还需要解析这些文本,判断要调用哪个函数并手动执行,再把结果拼回上下文。

这种方式在工程上存在明显问题:

如果不想在提示词里面做上述动作,那么就要在微调侧做投入,只不过当大模型原生支持 Tools(Function Calling)后,依旧是最优解:

工具调用能力被以标准化,包含明确的名称、功能描述和参数结构。模型不再“描述自己想做什么”,而是直接返回结构化的工具调用请求,应用或框架只需按约定执行对应函数并回传结果。

总结来说,提示词更适合描述目标和约束,而 Tools 更适合承载真实可执行的能力。当Agent开始具备实际行动能力时,使用 Tools 是唯一选择。

典型案例:旅行Agent

旅行规划是最经典的Tools案例,非常适合作为 Tools 的示例场景。

一个可用的旅行Agent,至少需要具备景点搜索、天气查询、酒店筛选、行程生成和路线规划等能力,这些能力都无法仅依赖模型内部知识完成,必须通过外部接口获取或计算。

在本示例中,我们先将旅行规划所需的能力拆解为多个独立工具,例如查询目的地景点信息、获取实时天气、根据预算筛选酒店、规划公共交通路线等。每个工具只负责一类明确的能力,返回结构化数据,不参与业务推理。

所有工具函数统一定义在 tools 目录中,与地图、地点、路线相关的能力通过高德开放平台 API 实现。工具本身只负责数据获取和计算,如何组合这些结果、如何生成最终的旅行方案,由模型在多轮推理中完成。

当工具准备好之后,还需要一套配套机制:将工具信息转换成模型可识别的 tools 参数;在模型返回工具调用指令时,执行对应的工具函数;并将工具执行结果回传给模型。这部分逻辑构成了旅行Agent与 Tools 之间的桥梁,下面将通过代码进行具体说明。

程序执行的时序图如下(其中的一种可能,具体的调用需要大模型决策执行) :

实现思路

为了构建这个旅行智能体,我们采用了模块化的代码设计,将“能力实现”与“智能决策”解耦。核心包含三个主要部分:

一、标准化的工具定义

我们没有为模型专门写一套复杂的配置文件,而是直接利用 Python 原生的语法特性来定义工具。在 TravelTools 类中,每一个方法(如 geocode, search_poi)都遵循以下规范:

class TravelTools: """ 旅游规划助手工具类 """ def init(self, amap_api_key: Optional[str] = None, request_delay: float = 0.2):

 self.amap_api_key = amap_api_key or os.getenv("AMAP_API_KEY") # ... 

async def estimate_travel_cost(self, city: str, days: int, hotel_level: str = "舒适",

 attractions: Optional[List[str]] = None) -> Dict[str, Any]: """ 估算旅游费用(不含往返交通) 使用场景: - 制定旅游预算 - 比较不同档次的旅游费用 - 规划旅游支出 Args: city: 旅游城市名称,如'西安'、'北京'、'上海' days: 旅游天数(含当天),如3表示2晚3天 hotel_level: 住宿档次,可选值: - '经济': 150元/晚(快捷酒店) - '舒适': 300元/晚(三星级酒店,默认) - '豪华': 500元/晚(四星级及以上) attractions: 计划游览的景点列表(可选),用于估算门票费用 如 ['兵马俑', '华清宫', '大雁塔'] Returns: 费用估算字典,包含: - city: 城市名称 - days: 旅游天数 - breakdown: 费用明细(住宿、餐饮、交通、门票) - total: 总费用 - tips: 温馨提示 """ # ... 具体实现逻辑 ... pass 

这种方式让开发者只需专注于编写标准的 Python 函数,而无需手动编写繁琐的 JSON Schema。

二、自动化注册与 Schema 生成

为了连接 Python 代码和大模型,我们实现了一个 ToolRegistry 工具类。它的核心职责是“翻译”和“管理”:

@dataclass class Tool: """工具定义数据模型""" name: str # 工具名称/函数名 description: str # 工具描述(用于大模型理解) parameters: Dict[str, Any] # 参数JSON Schema function: Callable # 实际的执行函数

def to_dict(self) -> dict:

 """转换为OpenAI Function Calling格式""" return { "type": "function", "function": { "name": self.name, "description": self.description, "parameters": self.parameters } } 

class ToolRegistry:

def register_from_class(self, cls, …):

 """从类中自动注册异步方法为工具""" # 利用 Python 的反射机制(Inspect模块) # 自动读取工具函数的签名和文档 # ... pass 

async def execute_tool(self, tool_call: ToolCall) -> ToolResult:

 """执行单个工具调用""" # ... pass 

三、ReAct循环

Agent的运行逻辑是一个经典的 While 循环,模拟了“观察-思考-行动”的过程:

# … 初始化 messages … while count < 15: count = count + 1

发送请求给大模型

result = await llm_client_with_tools(messages, tool_registry.get_tools()) assistant_message = result.choices[0].message

检查是否有工具调用

if assistant_message.tool_calls:

# 1. 添加 assistant 消息到历史 messages.append({ "role": "assistant", "content": assistant_message.content, "tool_calls": [...] }) # 2. 执行所有工具调用 for tool_call in assistant_message.tool_calls: # 执行工具 tool_result = await tool_registry.execute_tool(...) # 3. 添加工具结果到历史 messages.append({ "role": "tool", "tool_call_id": tool_call.id, "content": tool_result.content }) # 继续下一轮循环,将工具结果带给模型 continue 

else:

# 没有工具调用,输出最终回复 print(assistant_message.content) break 

通过这三层架构,我们将一个复杂的Agent拆解为:写好函数 -> 自动注册 -> 循环交互 的清晰流程,极大地降低了开发维护成本。

最后的效果如下:

在这个基础之下,我们再聊一聊上半年的热门词汇:MCP。

MCP(Model Context Protocol)是一种用于规范大模型与外部能力交互方式的协议。

它关注的不是某一个具体工具,而是如何以统一、标准的方式,把外部系统的能力、数据和上下文暴露给模型使用。

如果说 Tools 解决的是“模型如何调用一个函数”,那么 MCP 解决的是“模型如何与一个长期存在、可复用的能力服务交互”。

在 MCP 体系中,大模型并不直接面对零散的函数,而是通过协议连接到一个 MCP Server。这个 Server 可以对外提供多种能力,例如工具调用、资源读取、上下文查询等,模型通过 MCP Client 与之通信。

为什么会出现MCP

随着Agent应用越来越复杂,仅靠应用内定义的 Tools 会暴露出几个问题:

事实上在MCP之前,我们也会用各种工程手法去实现,其中一些经典实现已经接近了MCP,只不过自从官方提出来后,我们也就懒得折腾了;

MCP提供了一套标准化协议,把能力从单个应用中抽离,形成独立、可复用、可治理的服务层。通过 MCP:

至于要说他与Tools之间的关系,可以说一个是Agent的能力基石、另一个是Tools过多后的工程化管理实践结果:

在实践中,Tools 往往作为 MCP Server 内部能力的实现存在,但不再直接依赖某个应用。模型通过 MCP Client 调用能力时,只需关注“如何使用”,而不必关心实现细节,从而实现稳定、可复用且可扩展的Agent能力。

MCP的典型场景如下

接下来,我们来实际用用:

MCP Server 与 MCP Client

MCP 包含两个角色:

Server 负责能力的定义、执行和管理,Client 负责将模型的请求转换为 MCP 协议请求,并将返回结果交给模型:

MCP案例

依旧以之前的旅游Agent为例,MCP Server 端代码示例:

from typing import Dict, List, Optional, Any, Annotated from pydantic import Field from fastmcp import FastMCP

导入TravelTools类

from code.Function_Calling.tools import TravelTools mcp = FastMCP(name="旅游规划助手")

初始化TravelTools实例

travel_tools = TravelTools() @mcp.tool("get_current_weather", description="获取当前天气信息") async def get_current_weather(

 city: Annotated[str, Field(description="城市名称,如'西安'")], province: Annotated[str, Field(description="省份名称,如'陕西'")] 

) -> Dict[str, Any]:

try: weather = await travel_tools.get_weather(city, province) return weather.to_dict() except Exception as e: return {"error": str(e)} 

@mcp.tool("geocode", description="地理编码:将地址转换为经纬度坐标") async def geocode(

 address: Annotated[str, Field(description="地址或地点名称,如'兵马俑'、'大雁塔'、'西安市钟楼'")], city: Annotated[str, Field(description="城市名称(可选),用于限定搜索范围,如'西安'、'北京'")] = "" 

) -> Dict[str, Any]:

try: location = await travel_tools.geocode(address, city) return location.to_dict() except Exception as e: return {"error": str(e)} 

@mcp.tool("search_poi", description="搜索兴趣点(POI) - 关键词搜索") async def search_poi(

 keywords: Annotated[str, Field(description="搜索关键词,如'兵马俑'、'回民街'、'火锅'、'快捷酒店'")], city: Annotated[str, Field(description="城市名称,如'西安'、'北京'、'上海'")], types: Annotated[str, Field(description="POI类型筛选(可选),可选值:'景点'、'美食'、'酒店'、'交通'")] = "", page_size: Annotated[int, Field(description="返回结果数量,默认10,最大25")] = 10 

) -> List[Dict[str, Any]]:

try: pois = await travel_tools.search_poi(keywords, city, types, page_size) return [poi.to_dict() for poi in pois] except Exception as e: return [{"error": str(e)}] 

@mcp.tool("search_nearby", description="搜索周边POI - 基于位置的周边搜索") async def search_nearby(

 location: Annotated[str, Field(description="中心点坐标,格式为'经度,纬度',如'109.,34.'")], keywords: Annotated[str, Field(description="搜索关键词(可选),如'餐厅'、'便利店'、'ATM'")] = "", types: Annotated[str, Field(description="POI类型筛选(可选),可选值:'景点'、'美食'、'酒店'")] = "", radius: Annotated[int, Field(description="搜索半径,单位:米,默认1000米(1公里)")] = 1000, page_size: Annotated[int, Field(description="返回结果数量,默认5,最大25")] = 5 

) -> List[Dict[str, Any]]:

try: pois = await travel_tools.search_nearby(location, keywords, types, radius, page_size) return [poi.to_dict() for poi in pois] except Exception as e: return [{"error": str(e)}] 

@mcp.tool("route_planning", description="路线规划 - 多种出行方式的路线规划") async def route_planning(

 origin: Annotated[str, Field(description="起点坐标,格式为'经度,纬度'")], destination: Annotated[str, Field(description="终点坐标,格式为'经度,纬度'")], mode: Annotated[ str, Field(description="出行方式,可选值:'walking'(步行)、'driving'(驾车)、'transit'(公交/地铁)")] = "transit", city: Annotated[str, Field(description="城市名称,公交路线规划时必填,如'西安'、'北京'")] = "", origin_name: Annotated[str, Field(description="起点名称(可选),用于公交方案显示")] = "", destination_name: Annotated[str, Field(description="终点名称(可选),用于公交方案显示")] = "" 

) -> Dict[str, Any]:

try: route = await travel_tools.route_planning(origin, destination, mode, city, origin_name, destination_name) return route except Exception as e: return {"error": str(e)} 

@mcp.tool("get_weather_forecast", description="查询城市天气预报(未来3-4天)") async def get_weather_forecast(

 city: Annotated[str, Field(description="城市名称或城市编码(adcode),如'西安'、''")], province: Annotated[str, Field(description="省份名称,如'陕西'")], days: Annotated[int, Field(description="预报天数,默认4天(包含当天)")] = 4 

) -> Dict[str, Any]:

try: forecast = await travel_tools.get_weather_forecast(city, province, days) return forecast except Exception as e: return {"error": str(e)} 

@mcp.tool("estimate_travel_cost", description="估算旅游费用(不含往返交通)") def estimate_travel_cost(

 city: Annotated[str, Field(description="旅游城市名称,如'西安'、'北京'、'上海'")], days: Annotated[int, Field(description="旅游天数(含当天),如3表示2晚3天'")], hotel_level: Annotated[ str, Field(description="住宿档次,可选值:'经济'(150元/晚)、'舒适'(300元/晚)、'豪华'(500元/晚)")] = "舒适", attractions: Annotated[ Optional[List[str]], Field(description="计划游览的景点列表(可选),用于估算门票费用")] = None 

) -> Dict[str, Any]:

try: cost = travel_tools.estimate_travel_cost(city, days, hotel_level, attractions) return cost except Exception as e: return {"error": str(e)} 

@mcp.tool("get_attraction_info", description="获取景点详细信息") async def get_attraction_info(

 attraction_name: Annotated[str, Field(description="景点名称,如'兵马俑'、'华清宫'、'大雁塔'")], city: Annotated[str, Field(description="城市名称,如'西安'、'北京'")] 

) -> Dict[str, Any]:

try: info = await travel_tools.get_attraction_info(attraction_name, city) return info except Exception as e: return {"error": str(e)} 

if name == ‘main’:

mcp.run(transport="http", port=8001) 

MCP Client 端代码示例:

import asyncio import json import os from openai import OpenAI from fastmcp import Client from code.Working_with_LLMs.llm_client import llm_client_with_tools

系统提示词

SYSTEM_PROMPT = """你是一个专业的旅游规划助手。当用户提出旅游规划需求时,请:

  1. 理解需求:确认目的地、天数、预算、出行人数、特殊偏好
  2. 使用工具搜索:
    • 使用 search_poi 搜索热门景点
    • 使用 get_attraction_info 获取景点详情
    • 使用 estimate_travel_cost 估算费用
    • 使用 route_planning 规划路线
  3. 输出格式: 按天输出详细行程,每天包含:
    • 上午/下午/晚上的景点安排
    • 餐饮推荐
    • 景点间交通方式和时间
    • 当日预估花费
  4. 费用把控:根据用户预算合理分配费用
  5. 贴心提醒:提供穿着建议、必带物品、注意事项

请确保使用工具获取最新信息,并给出具体的行程安排。"""

async def chat_with_mcp(user_input: str):

client = OpenAI(api_key=os.getenv("DEEPSEEK_API_KEY"), base_url="https://api.deepseek.com") # 连接到你的服务器 mcpClient = Client("http://127.0.0.1:8001/mcp") # 如果是HTTP服务器,这里放URL await mcpClient.__aenter__() # 列出可用工具 mcp_tools = await mcpClient.list_tools() print("可用工具:", mcp_tools) llm_tools = [] for tool in mcp_tools: llm_tools.append({ "type": "function", "function": { "name": tool.name, "description": tool.description, "parameters": tool.inputSchema, }, }) # 加载工具 # 第一次调用,让模型自己判断是否使用工具 messages = [{"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": user_input}] count = 0 while count < 15: count = count + 1 result = await llm_client_with_tools(messages, llm_tools) print(result) assistant_message = result.choices[0].message content = assistant_message.content or "" # 检查是否有工具调用 if assistant_message.tool_calls: # 构建tool_calls记录 tool_calls_data = [ { "id": tc.id, "type": tc.type, "function": { "name": tc.function.name, "arguments": tc.function.arguments } } for tc in assistant_message.tool_calls ] # 添加到消息历史 messages.append({ "role": "assistant", "content": content, "tool_calls": tool_calls_data }) # 执行所有工具调用 for tool_call in assistant_message.tool_calls: tool_name = tool_call.function.name print("正在执行工具:", tool_name) try: arguments = json.loads(tool_call.function.arguments) print("工具参数:", arguments) except json.JSONDecodeError: arguments = {} # 执行工具 tool_result = await mcpClient.call_tool(tool_name, arguments) print("工具结果:", tool_result) # 添加工具结果 messages.append({ "role": "tool", "tool_call_id": tool_call.id, "content": json.dumps(tool_result.structured_content, ensure_ascii=False) }) continue else: count = 15 print(content) 

if name == ‘main’:

asyncio.run(chat_with_mcp("西安2日游")) 

虽然MCP解决了多Agent对Tools的调用解耦,但Agent的老大难问题依旧没被解决,核心就两点:

在这个基础下,我们依旧有很多的工程手段去做优化:

只不过这个工程优化点与MCP一样,很快被官方以标准化的方式做掉了:Anthropic 官方文档给出了 Skills 的解决方案,他包含三个层级,从抽象到具体:

虽然Skills首先由Claude推出,但现在几乎成为了事实上的标准:

深度解析

Claude Skills 设计遵循了一个非常重要的原则:Progressive Disclosure,渐进式批露

分阶段、按需加载信息,而不是在任务开始时就将所有内容全部塞入上下文窗口中。整个加载过程分为三个层次,对应上面的三要素:

第一层:元数据(始终加载)

Claude 在启动时会扫描所有已安装的 Skills,并加载这些元数据,将其纳入系统提示(System Prompt)中。作用:

— name: douyin-summary

description: 抖音视频总结助手。当用户提供抖音(douyin.com 或 v.douyin.com)视频链接并请求总结、获取文案或了解视频内容时,使用此技能。通过调用 Coze API 工作流获取视频的转录文本或文案,然后为用户提供智能化的内容总结。

第二层:核心指令(触发时加载)

当用户的请求与某个 skill 的描述相匹配时,Claude 会通过 bash 从文件系统中读取对应的 SKILL.md 文件,并将其完整内容加载进当前对话上下文:

# 抖音视频总结助手 此技能用于获取和总结抖音视频的内容。

工作流程

当用户提供抖音链接时:

  1. 识别抖音链接: 检测用户输入中的 douyin.com 或 v.douyin.com 链接
  2. 调用脚本获取内容: 使用 scripts/fetch_douyin.py 获取视频转录/文案 python3 ./scripts/fetch_douyin.py
    1. 总结内容: 基于获取的文本内容,提取核心观点、关键信息或有趣之处
  3. 友好输出: 以简洁易懂的方式呈现给用户

    第三层:代码与资源(按需加载)

    一个复杂的 skill 可能包含多个文件,形成一个完整的知识库。skill 可以将这些资源与指令一起打包,实现完整的任务闭环。

    通过 元数据 → 指令 → 代码与资源 这三层结构,一个 skill 不仅能被 Claude 正确识别和触发,还能真正完成从“理解需求”到“执行任务”的完整闭环:

    └── skill-name/ # 技能根目录 ├── meta.json # [路由层] 告诉模型这个技能是干嘛的 ├── skill.md # [逻辑层] 包含了 System Prompt 和 SOP └── scripts/ # [执行层] 实际干活的 Python/Bash 脚本

接下来给个具体的案例:

Skills的安装与使用

在Claude Code中,你可以通过以下两种方式使用Skills:

方法一:使用官方技能市场(推荐)

# 添加官方技能库 /plugin marketplace add anthropics/skills

浏览可用技能

/plugin list

安装文档处理技能

/plugin install document-skills@anthropic-agent-skills

安装完成后,你可以直接询问Claude有哪些技能:

方法二:手动创建自定义技能

如果你想让自己定义一个skills,比如“自动总结抖音视频”,你可以自己编写一个 Skill。 原理非常简单:

目录结构看起来是这样的:

1.创建技能目录:

mkdir -p ~/.claude/skills/douyin-summary 

1.编写SKILL.md:

— name: douyin-summary

description: 抖音视频总结助手。当用户提供抖音视频链接时,自动调用此技能获取文案并总结。

抖音视频总结助手

工作流程

  1. 识别用户输入中的douyin.com链接
  2. 调用scripts/fetch_douyin.py获取视频文案
  3. 提取核心观点并结构化输出

    3.实际使用:

    配置好后,可以查看技能是否安装成功。老版本的Claude code 需要重启才能看到新安装的技能,新版本已经不需要重启了:

    使用Claude Skills提取抖音视频内容的效果如下图所示:

    核心理念

    Skills 是一种模块化、可复用的能力单元。它不仅仅包含“能做什么”(Function Call),还包含“怎么做”(Prompt/SOP)以及“用什么做”(Code/Resources),其核心为:

    在 Claude 的定义中,skill 表现为一个标准的文件系统目录,包含了描述、指令和执行脚本:

    Skills与Function Calling

    很多同学容易混淆这两个概念。我们可以通过一个对比表来看清它们的本质区别:

    一句话总结:Tools 是能力的低层接口(API);Skills 是任务的高级模板(SOP + Tools)。

    一个 skill 内部通常会调用一个或多个 Tools 来完成具体工作,但它更强调流程编排和知识注入。

    优势说明

    Skills 最初是由 Anthropic 在 Claude 生态中形式化的概念,但它所代表的“能力即文件”(Capability as Files)的设计思想,已经超越了特定平台的限制,成为构建复杂 Agent 的一种通用**实践。

    所以,各个基模平台都在跟进,这里的原因是:

    一、动态上下文管理

    传统的 Agent 开发往往将所有能力的 Prompt 一次性写入 System Prompt,这会导致:

    Skills 模式的核心在于按需加载:只有当模型判定需要处理特定任务(如“查询天气”)时,系统才会读取对应的 skill.md 并注入上下文。

    这种渐进式披露机制让 Agent 能以较小的 Context 承载近乎无限的能力库。

    只不过这里的说法依旧有些夸张,当Tools基数达到一个量时候,又会有skill过多的问题,这时候可能会产生基于Skills的“Skills技术”

    二、标准化的能力封装

    skill是三位一体(Meta + Prompt + Code)的结构,让 Agent 的能力变得像代码库一样,可以被:

    三、模型无关性

    Skills 本质上只是一种文件结构约定。

    只要我们编写代码去解析这些文件,并根据 meta.json 的描述与大模型交互,就可以让 DeepSeek、OpenAI 或任何具备 Tool Calling 能力的模型 拥有“加载 Skills”的能力。

    这也是说其他模型都在跟进这种工程优化能力的原因,其实他们不跟进也没关系,我们自己做实现就好了,我们这里说干就干:

    在DeepSeek上实现Skills引擎

    首先,我们需要读取本地文件夹,把 meta.json 转换成大模型能看懂的 Tool Definition:

    def load_skills_from_meta(): """扫描目录,从 meta.json 加载技能元数据"""

    示例:假设我们读取到了 simple_weather_skill

    with open("./simple_weather_skill/meta.json", "r", encoding="utf-8") as f: meta = json.load(f)

构造兼容 OpenAI/DeepSeek 格式的工具定义

return [{

"type": "function", "function": { "name": meta["name"], "description": meta["description"], "parameters": { "type": "object", "properties": {}, "required": [] }, }, 

}]

当用户提问时,我们把加载好的 skills 列表发给 DeepSeek,利用它的 Function Calling 能力来判断"需要使用哪个 Skill",这也是最核心的意图识别

# 1. 准备工具列表 skills = load_skills_from_meta()

2. 发送请求给 DeepSeek,让其选择

messages = [{"role": "user", "content": user_input}] result = client.chat.completions.create(

model="deepseek-chat", messages=messages, tools=skills # 关键点:把 Skill 描述当作 Tools 传进去 

)

然后就是上下文注入了,如果模型决定调用某个 skill(例如 weather_skill),我们不执行它,而是读取该 skill 的 skill.md,把它追加到 System Prompt 中:

# 检查模型是否想调用 Skill if result.choices[0].message.tool_calls:

skill_name = result.choices[0].message.tool_calls[0].function.name print(f"模型命中技能: {skill_name}") # 读取对应的 Prompt 模板 with open(f"./{skill_name}/skill.md", "r") as f: skill_prompt = f.read() # 【核心魔法】:构建一个新的对话上下文,使其立即拥有该技能的知识 skill_messages = [ {"role": "system", "content": skill_prompt}, # 注入技能 Prompt {"role": "user", "content": user_input} # 重放用户问题 ] # 此时,DeepSeek 已经变身为“天气专家”,准备好执行具体脚本了 

最后是执行,在注入了 skill.md 的新会话中,模型会根据 Prompt 的指示,去调用 scripts/ 下真正的 Python 脚本(通常通过 execute_script 这样一个通用工具来实现),最终返回结果:

def execute_script(script_path, args=None):

"""一个通用的脚本执行器工具""" cmd = ["python", script_path] + (args or []) res = subprocess.run(cmd, capture_output=True, text=True) return res.stdout 

通过这套机制,我们就在 DeepSeek 上完美模拟了 Claude 的 Skills 流程:路由 -> 加载 Prompt -> 执行脚本。

接下来是基于记忆系统、ReAct框架与一个小红书爆款实践案例。

真正做过复杂AI项目的同学都会理解,在AI应用层最难的其实是:数据如何与AI做交互

比如我看过几个生产级复杂AI项目,代码量就1万行左右,但所依赖知识库却大的惊人,这里伴有强烈的非对称性,其结果是复杂AI项目,80%的时间都在处理数据问题!

因此,我们会对于复杂AI项目中AI如何与数据交互的范式(**实践)是特别关注的,也就是在这个基础下出了很多名词:短期记忆、长期记忆、语义记忆、情景记忆…

所有这一切,构成了Agent的记忆系统。如果说 LLM是 Agent 的大脑,负责思考和决策;那么 Memory 就是 Agent 的海马体,负责记录经验和知识。

LLM 的核心特性是:无状态。对模型来说,每一次调用都是全新的开始。无论你们之前聊得多么热火朝天,一旦开启新的一轮对话,它就把前面发生的一切忘得干干净净。

为了让 Agent 具有“记忆”,就需要构建一套外部记忆系统。这套系统不仅仅是简单的“把聊天记录存下来”,它需要解决三个核心工程挑战:

在工程语境下,Memory 指的是模型在当前输入之外,仍然能够访问和使用的信息集合。这些信息可能来自历史对话、外部存储或系统内部状态,但核心目标只有一个:为当前推理提供必要的上下文补充。

从内容性质上看,Agent 中的记忆通常可以分为三类:

这是一种逻辑分类,并不等同于具体的存储或实现方式。真正落地时,这些记忆往往会以不同工程形态存在。

最后要注意的是:记忆系统是整个Agent最复杂、金贵的部分,我们这里只会给出最简单的介绍,系统性的介绍还是得上课

在实际系统中,很少直接实现抽象的“情景记忆”或“语义记忆”,而是通过几种常见的工程模式来承载它们:

上下文记忆

上下文记忆是最基础的实现方式,其是将历史对话原样拼接到当前 Prompt 中,一并发送给模型,模型通过看到之前的对话内容保持语义连贯性。

这种方式实现成本极低,适合原型验证或短对话场景,但受限于模型的 Token 上限,对话越长成本越高,也无法支持跨会话或长期记忆。从本质上看,它是一种短期、一次性的情景记忆。

滑动窗口记忆

滑动窗口记忆是在上下文记忆基础上的一种约束策略,只保留最近固定轮数的对话,其余内容直接丢弃。它解决的不是“记忆能力”问题,而是“Token 成本控制”问题。

在工程上可以理解为:情景记忆的生命周期管理机制。它适合上下文有效期明确、业务流程较短的场景,但一旦关键信息被滑出窗口,就会永久丢失。

摘要记忆

摘要记忆通过调用模型对历史对话进行压缩,将大量情景记忆转换为一段简要描述,并在后续对话中使用该摘要替代原始内容。

这种方式在成本和上下文长度上具有明显优势,但摘要不可避免造成信息丢失,质量高度依赖模型能力。因此它更适合保留“整体脉络”,而不适合依赖精确细节的场景。

从记忆类型上看,摘要记忆本质上是将情景记忆转化为低精度的语义记忆。

向量记忆

向量记忆是一种典型的长期记忆实现方式。其核心做法是将对话内容、经验或用户偏好向量化后存入向量数据库,在需要时通过语义相似度检索相关内容。

这种方式不受对话长度限制,适合长期知识积累和跨会话记忆,但检索结果是“语义相似”而非“精确匹配”,实现复杂度高于前几种方式。它是当前 Agent 系统中最常见的语义记忆工程实现。

记忆存储

有些信息是需要存储起来的,这有很多目的:

根据需求和项目复杂度,通常涉及以下几类存储:

我们后续的存储方式是SQLite + ChromaDB

关键连接:RAG

SQLite + ChromaDB 的混合存储方案,相当于给 Agent 配备了硬盘。

RAG在Agent记忆系统中扮演了关键角色。它的核心职责是在有限的上下文窗口内,精准地加载最相关的长期记忆。对于 Memory 模块,RAG 不仅仅是“搜索”,它包含了一个完整的记忆调度流程:

通过 RAG,解决了 Agent 记不住(无状态)和记不下(Token 限制)的矛盾。

接下来是具体实现:

为了实现上述目标:解决Agent记不住、存不下的问题,我们设计了一个分层的 Memory 架构:

整体流程如下:

在写代码之前,最后需要明确的是“怎么管”这些数据,我们需要制定以下核心策略:

一、双写一致性策略

我们既有 SQLite 又有 ChromaDB,如何保证它们数据不打架?

二、混合检索策略

为了避免检索出完全不相关的内容,我们需要给检索加“围栏”:

三、生命周期管理

滑动窗口:L1 热记忆只保留最近 10-20 轮。

持久化:L2 和 L3 必须落盘,确保服务重启后“失忆”不会发生。

衰减(可选):虽然本期暂不实现复杂的艾宾浩斯遗忘曲线,但在设计表结构时,我们预留了 importance_score(重要性评分)和 last_accessed_at(最后访问时间),为未来实现“不常用的记忆自动淡化”打好基础。

至此就可以开始代码实现了:

基于前面的理论基础,我们实现了一个完整的记忆系统,包含数据存储、向量检索、RAG 增强和 LLM 交互等核心功能。

整个系统采用分层架构,从底层到上层依次为:

应用层: chat_with_memory.py (LLM 交互入口) ↓ 工具层: memory_tools.py (高级封装) ↓ 服务层: memory_service.py (业务逻辑) ↓ 存储层: vector_store.py + models.py (数据访问) ↓ 数据层: ChromaDB + SQLite (实际存储) 

这里的设计原则是:

数据模型设计

一、users表(用户)

class User(Base):

id = Column(Integer, primary_key=True) username = Column(String(100), unique=True, nullable=False) email = Column(String(255), unique=True, nullable=True) created_at = Column(DateTime, default=datetime.utcnow) updated_at = Column(DateTime, default=datetime.utcnow) 

二、sessions 表(会话)

class Session(Base):

id = Column(Integer, primary_key=True) session_id = Column(String(100), unique=True, nullable=False) user_id = Column(Integer, ForeignKey('users.id')) title = Column(String(255), nullable=True) context_window = Column(Integer, default=10) # 上下文窗口大小 is_active = Column(Boolean, default=True) created_at = Column(DateTime, default=datetime.utcnow) 

三、memories 表(记忆)

class Memory(Base):

id = Column(Integer, primary_key=True) user_id = Column(Integer, ForeignKey('users.id')) session_id = Column(Integer, ForeignKey('sessions.id'), nullable=True) memory_type = Column(String(50)) # episodic/semantic/procedural content = Column(Text, nullable=False) metadata = Column(JSON, nullable=True) vector_id = Column(String(100)) # ChromaDB 向量ID importance_score = Column(Float, default=0.5) # 重要性评分 access_count = Column(Integer, default=0) # 访问次数 last_accessed_at = Column(DateTime, nullable=True) is_deleted = Column(Boolean, default=False) # 软删除 

设计要点:

向量存储实现

使用 ChromaDB 作为向量数据库,核心代码如下:

class VectorStore: def init(self, persist_directory: str = "./chat_chroma.db"): self.client = chromadb.Client(Settings( persist_directory=persist_directory, anonymized_telemetry=False )) self.collection = self.client.get_or_create_collection( name="memory_collection" )def add_memory(self, content: str, metadata: Dict = None) -> str: """添加记忆到向量库""" memory_id = str(uuid.uuid4()) metadata = metadata or {} metadata[‘created_at’] = datetime.utcnow().isoformat() self.collection.add( ids=[memory_id], documents=[content], metadatas=[metadata] ) return memory_iddef search_memories(self, query: str, n_results: int = 5, filter_metadata: Dict = None) -> Dict: """语义搜索记忆""" results = self.collection.query( query_texts=[query], n_results=n_results, where=filter_metadata ) return { ‘ids’: results[‘ids’][0], ‘documents’: results[‘documents’][0], ‘metadatas’: results[‘metadatas’][0], ‘distances’: results[‘distances’][0] } 

关键特性:

记忆服务层

MemoryService 整合了数据库和向量库,提供统一的记忆管理接口。核心方法包括添加记忆(双写机制):

def add_memory(self, user_id: int, content: str,

 memory_type: str = "episodic", session_id: int = None, metadata: Dict = None, importance_score: float = 0.5) -> Memory:"""添加记忆(同时保存到数据库和向量库)"""db = self._get_db_session()try: # 1. 创建数据库记录 memory = Memory( user_id=user_id, session_id=session_id, memory_type=memory_type, content=content, metadata=metadata or {}, importance_score=importance_score ) db.add(memory) db.commit() db.refresh(memory) # 2. 添加到向量库 vector_metadata = { 'user_id': user_id, 'memory_type': memory_type, 'memory_db_id': memory.id } vector_id = self.vector_store.add_memory( content=content, metadata=vector_metadata ) # 3. 更新向量ID memory.vector_id = vector_id db.commit() return memoryfinally: db.close() 

搜索记忆(语义检索 + 数据库查询):

def search_memories(self, query: str, user_id: int = None,

 memory_type: str = None, n_results: int = 5) -> List[Dict]:"""搜索记忆"""# 1. 向量搜索filter_metadata = {}if user_id: filter_metadata['user_id'] = user_idif memory_type: filter_metadata['memory_type'] = memory_typeresults = self.vector_store.search_memories( query=query, n_results=n_results * 2, filter_metadata=filter_metadata if filter_metadata else None)# 2. 整合数据库信息memories = []db = self._get_db_session()try: for i, vector_id in enumerate(results['ids']): metadata = results['metadatas'][i] memory_db_id = metadata.get('memory_db_id') if memory_db_id: memory = db.query(Memory).filter( Memory.id == memory_db_id, Memory.is_deleted == False ).first() if memory: # 更新访问统计 memory.access_count += 1 memory.last_accessed_at = datetime.utcnow() memories.append({ 'id': memory.id, 'content': memory.content, 'memory_type': memory.memory_type, 'importance_score': memory.importance_score, 'distance': results['distances'][i], 'created_at': memory.created_at.isoformat() }) if len(memories) >= n_results: break db.commit() return memoriesfinally: db.close() 

核心逻辑:

高级工具层

MemoryTools 提供面向应用的高级接口,最核心的是 RAG 检索功能:

def retrieve_for_query(self, user_id: int, query: str,

 session_id: int = None, include_context_window: bool = True, max_results: int = 5) -> Dict:"""RAG 检索:整合短期和长期记忆"""result = { 'context_window': [], 'relevant_memories': []}# 1. 获取上下文窗口(短期记忆)if include_context_window and session_id: context_memories = self.service.get_context_window(session_id) result['context_window'] = [ {'content': m.content, 'created_at': m.created_at.isoformat()} for m in context_memories ]# 2. 语义搜索(长期记忆)relevant = self.service.search_memories( query=query, user_id=user_id, n_results=max_results)result['relevant_memories'] = relevantreturn result 

RAG 实现要点:

ChatWithMemory 类整合了记忆系统和 LLM 调用:

class ChatWithMemory:

def chat(self, user_input: str, use_rag: bool = True) -> str: """与 LLM 聊天(带记忆)""" messages = [{"role": "system", "content": self.system_prompt}] if use_rag: # RAG 检索 rag_result = self.memory.retrieve_for_query( user_id=self.user_id, query=user_input, session_id=self.session_id, include_context_window=True, max_results=3 ) # 添加相关长期记忆 if rag_result['relevant_memories']: memory_context = "相关历史信息: 

" for mem in rag_result[‘relevant_memories’][:3]: memory_context += f"- {mem[‘content’]} " messages.append({"role": "system", "content": memory_context}) # 添加上下文窗口 for ctx in rag_result[‘context_window’]: # 解析对话内容 if"User:"in ctx[‘content’] and "Assistant:"in ctx[‘content’]: parts = ctx[‘content’].split(" Assistant:") user_part = parts[0].replace("User:", "").strip() assistant_part = parts[1].strip() messages.append({"role": "user", "content": user_part}) messages.append({"role": "assistant", "content": assistant_part}) # 当前输入 messages.append({"role": "user", "content": user_input}) # 调用 LLM response = call_llm(messages=messages) assistant_message = response.choices[0].message.content # 保存对话记忆 self.memory.add_conversation_memory( user_id=self.user_id, session_id=self.session_id, user_message=user_input, assistant_message=assistant_message ) return assistant_message

完整流程:

使用示例
# 基础使用 from memory_tools import MemoryTools# 初始化memory = MemoryTools()# 创建用户和会话user_id = memory.create_user("alice")session_id = memory.create_session(user_id, "session-001")# 添加对话记忆memory.add_conversation_memory( user_id=user_id, session_id=session_id, user_message="我喜欢Python编程", assistant_message="Python是一门很棒的语言!")# 搜索记忆results = memory.search_memories( user_id=user_id, query="编程语言", n_results=5) 

带记忆的聊天:

# 运行交互式聊天 python chat_with_memory.py

示例对话

你: 我叫Alice,喜欢编程 助手: 你好Alice!很高兴认识你。编程是一个很有趣的领域…

你: 我之前说过我叫什么名字? 助手: 你之前说你叫Alice。

系统会自动:

小结

这个记忆系统的实现展示了如何将理论概念转化为可用的工程系统:

通过这套系统,Agent 可以:

这正是 Memory 在 Agent 系统中的核心价值所在。

在了解记忆系统后,便可进入ReAct框架模块了,说白了就是如何循环生成任务规划。

任务规划(Planning)是 Agent 系统中的核心控制模块,负责将复杂的自然语言指令转化为结构化的可执行步骤。

它连接了LLM的推理能力与工具Tools的执行能力,通过“先规划、后执行”的模式,解决直接生成执行动作带来的不确定性与幻觉问题。

PS:只不过这个想法很好,事实上在Skills之前,不稳定性挺高的

ReAct的核心模块包含三个组件:

这个东西其实复杂度不高,就一个循环罢了,我们直接上代码实现:

数据模型定义

系统通过 Plan 和 PlanStep 类定义标准的数据结构。每个步骤包含唯一的 ID、工具名称、参数以及执行状态。

@dataclass class PlanStep:

id: str tool: str args: Dict[str, Any] why: str = "" status: Literal["pending", "running", "success", "failed", "skipped"] = "pending" output: Any = None error: str = "" 

@dataclass class Plan:

goal: str steps: List[PlanStep] = field(default_factory=list) 

规划器设计

规划器通过 System Prompt 强制 LLM 输出 JSON 格式,并注入当前可用的工具描述。

为了实现步骤间的数据传递,我们定义了模板语法 {{steps.step_id.output.field}},允许后续步骤引用前序步骤的输出:

system_prompt = f""" 你是一个任务规划器(Planner)。你必须只输出 JSON,不要输出任何解释文字。 目标:把用户的需求拆成若干步,每一步要调用一个工具(tool),并提供参数(args)。

关键规则——数据依赖: 如果某一步的 args 需要依赖前面步骤的输出结果,请务必使用模板语法 {{{{steps.前面的步骤ID.output.字段名}}}}。 不要自己捏造数据,而是引用前序步骤的输出。

例如:

  1. step1 (id="s1") 调用 geocode 返回 {{"location": "116.40,39.90"}}
  2. step2 (id="s2") 需要用到这个坐标,args 应该写: {{"location": "{{{{steps.s1.output.location}}}}"}}

你必须输出的 JSON 结构: {{ "goal": "…", "steps": [

{{ "id": "step1", "tool": "tool_name", "args": {{ ... }} }} 

] }} """

执行器与参数解析

执行器在调用工具前,会先扫描参数中的模板占位符,并从当前的执行上下文(Context)中提取真实值进行替换。这种机制实现了模型规划与具体数据处理的解耦:

async def _run_step(self, step: PlanStep, mcp_client: Client, context: Dict[str, Any]) -> None: step.status = "running"try: # 解析参数中的 {{…}} 模板,替换为上下文中的真实数据 resolved_args = resolve_templates(step.args, context) # 调用 MCP 工具 result = await mcp_client.call_tool(step.tool, resolved_args) # 记录输出结果 step.output = getattr(result, "structured_content", None) or result.content step.status = "success" # 写入上下文,供后续步骤引用 context["steps"][step.id] = { "status": step.status, "output": step.output }except Exception as e: step.status = "failed" step.error = str(e) 
运行示例

项目包含一个 CLI 演示程序,用于展示完整的规划与执行过程。

启动 MCP 服务器: 确保 MCP 服务已启动,提供基础工具能力。

uv run python -m code.MCP.mcp_server 

运行 Planning CLI:

uv run python -m code.Planning.demo_planning_cli.py 

执行效果示例:

看吧,是不是确实很简单,最后我们用一个真实案例将这些知识点串起来:

先上一些效果图:

在Agent基础能力(工具调用、记忆、规划)逐步完善后,我们开始探索更具实用价值的集成场景。

社交媒体内容创作与发布是一个典型的重复性工作流程,适合通过自动化提升效率。我们今天用Agent来实现从小红书主题输入到平台发布的全流程自动化。

该Agent接收自然语言指令(如"生成一篇关于露营装备的推荐笔记"),自动完成内容创作、配图生成、平台登录、内容填充及发布的全过程。其整体流程如下:

我们在Coze平台构建了一个多节点协同的工作流,专门生产符合小红书平台特性的结构化内容包。该工作流不是简单的文本生成,而是遵循"爆款公式"的标准化生产线:

这里有几个核心点:

一、主题分析与结构规划节点

这里也可以不用Coze,但是他里面已经有了很多工作流挺好用的。

用户输入爆文主题,可以是一个新闻稿,也可以是引发思考的一句话,比如下文中的“Gemini3 是目前最强 AI 吗?”

随后,调用LLM分析主题关键词,利用大模型节点中的新闻搜索技能,收集到的信息整理后输出文案;

最终输出:内容大纲框架,包含标题变体、情绪钩子、分段逻辑、互动结尾设计

二、文案生成与图片提示词生成

三、结构化打包节点

async function main({ params }: Args): Promise { // 构建输出对象 const ret = { fields: [ { fields: { "小红书文案": params.content, "小红书标题": params.title, "小红书图文地址": params.imageUrl }, record_id: params.record_id } ]}; return ret;} 
实现步骤拆解

当我们有一段文字素材、新闻稿,甚至只是一句话灵感时,如何快速生成适用于抖音、小红书、公众号这三个主流自媒体平台的内容?

这三个平台用户活跃度高,风格各异,传统手动改写耗时耗力,因此我们希望通过自动化方式,实现“一次输入,多平台适配”。

飞书多维表格在这里承担了“内容数据库”的角色。为了支持后续自动化流程,首先要设计清晰的字段结构。

先建立一个线索表,用于存放初始内容素材,如文字素材、新闻稿,或简单的一句话灵感;

再建立文章产出表,用于存放扣子写回的文章数据,包括图片的URL,标题,文章

现在飞书表格设计好以后,根据飞书的表格字段,去设计扣子工作流

Coze工作流实现

在Coze工作流中,为大模型节点添加了“新闻识别”技能,使其能够根据简短的一句话自动生成完整的文章。

当输入内容较少时,大模型会自动提取关键词进行新闻搜索,获取相关信息作为语料,进而辅助文章生成:

提示词的设计对模型理解任务至关重要,推荐使用Markdown格式进行结构化组织,明确指定角色、任务、技能、要求和输出格式等。这种结构化的方式能够帮助大模型更准确地把握任务目标,提升生成内容的质量。

选用的是豆包1.6深度思考模型,该模型在内容生成质量上表现更优,但运行速度相对较慢。若提示词较为复杂,且对生成时长要求不高,推荐使用思考模型以获得更优质的结果:

考虑到公众号、小红书和抖音在内容形式上的显著差异,我对不同平台进行了独立设计:

三个平台的内容生成流程执行完毕后,结果会自动写回到飞书表格中,便于后续管理与使用。

注意:工作流偶尔可能因思考模型运行超时而报错,建议适当延长等待时间或配置错误处理机制,此处不展开详述。

以下为完整流程示例,输入内容为:“Gemini3 是目前最强 AI 吗?”

下面是飞书中收集到的信息:

现在开始飞书对接扣子,扣子工作流制作完成后,需要发版,发版后的工作流就是一个API,可以提供给外部引用调用。

飞书对接扣子

在飞书平台配置自动化流程时,选择飞书字段触发器为按钮。用户可以通过点击按钮发送HTTP请求,从而触发已经封装好的扣子工作流,实现流程的自动化启动:

完成触发器配置后,在飞书的自动化流程编辑器中进一步设置按钮操作。下图展示了飞书与扣子工作流对接流程:

在工作流部分处理结束后,在生成内容这块问题不大了,但Agent并不具备发布功能,所以这里还需要操作浏览器,浏览器操作与Computer-Use也是非常基础的Agent工具了。

与传统的基于DOM元素定位的自动化工具不同,我们采用了一种更接近人类操作方式的方法:让LLM理解屏幕内容并做出决策。

核心原理是通过计算机视觉分析浏览器界面,结合页面HTML结构,让LLM像真人一样"看到"页面并决定下一步操作。核心代码实现如下:

这里的 Agent,其实是 browser-use 工具里封装好的一个内置 Agent。它运行时需要三个核心参数:llm、task、browser:

回顾之前开发的旅游 Agent,本质结构和这个内置 Agent 完全一致:都是 LLM + Tools 的组合。

区别在于现在我们在自己的 Agent 内部,再去调用一个封装好的 Agent,这就形成了一个 多 Agent 协作 的示例。执行流程解析:

小结

至此,这个案例就差不多了,使用Browser-Use解决了一些问题,只不过实践下来,这东西没那么稳定,大家做demo可以,如果生产环境还是上影刀算了

文章内容很长了,这里不赘述了,本次课程核心有两个案例,第一个几乎将Agent的关键词Function Calling、MCP、Skills用完了的旅游助手

第二个是新增的一个自媒体相关工具,会用一些奇奇怪怪的工具,至于如何将他们串起来,大家多多感受即可。

总之,都是Agent……

小讯
上一篇 2026-04-16 13:13
下一篇 2026-04-16 13:11

相关推荐

版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容,请联系我们,一经查实,本站将立刻删除。
如需转载请保留出处:https://51itzy.com/kjqy/262643.html