Walter's Blog

主题

从零开始构建 Agent 和 Skill ​

前置知识 ​

动手之前，先确认你理解了以下概念：

必须理解 ​

概念是什么为什么需要
LLM API大模型的调用接口Agent 的"大脑"靠它思考
Function CallingLLM 原生支持的函数调用机制让 LLM 知道"什么时候该调用什么工具"
JSON Schema描述 JSON 数据结构的规范定义 Skill/Tool 的输入参数格式
System Prompt给 LLM 的角色设定和行为约束定义 Agent 的人设、能力和边界
ReAct 循环Reasoning + Acting 交替执行的模式Agent 的基本运行方式

建议了解 ​

概念是什么为什么需要
MCPModel Context Protocol，工具调用的标准协议做跨平台、可复用的 Skill 时需要
向量数据库存储和检索语义相似度的数据库给 Agent 加长期记忆（RAG）时需要
Async PythonPython 异步编程高并发场景下的 Agent 性能优化

需要的工具 ​

工具用途
Python 3.10+开发语言
uv / pip包管理
LLM SDK（见下方选择指南）调用大模型
Kubernetes 客户端（以 K8s 场景为例）操作 K8s 集群

LLM 选择指南 ​

::: important 不是只有本地部署！任何支持 Function Calling 的大模型都可以，云端 API 或本地部署都行。 :::

方式示例优点缺点
云端 APIOpenAI、Anthropic、通义千问、DeepSeek开箱即用、效果好、无需 GPU按量付费、数据经过第三方
本地部署Ollama + Llama3、vLLM + Qwen数据不出本机、无使用费需要显卡、效果取决于模型大小
私有化部署企业内部部署的大模型服务兼顾安全和效果需要运维基础设施

关键前提：模型必须支持 Function Calling。大部分主流模型都已支持：

模型Function Calling 支持调用方式
OpenAI (GPT-4o / GPT-4o-mini)✅ 原生支持openai SDK
Anthropic (Claude Sonnet/Opus)✅ 原生支持anthropic SDK
DeepSeek✅ 原生支持兼容 OpenAI SDK
通义千问 (Qwen)✅ 原生支持兼容 OpenAI SDK
Ollama 本地模型⚠️ 部分模型支持兼容 OpenAI SDK
智谱 (GLM)✅ 原生支持兼容 OpenAI SDK

为什么很多模型都兼容 OpenAI SDK？

因为 OpenAI 的 API 格式成了事实标准，大部分模型厂商和本地部署工具（Ollama、vLLM、LiteLLM）都提供了兼容接口，只需改 base_url 就能切换模型。SDK 和 Agent 代码完全不用改，只换 base_url、api_key 和 model 三个值。

下面是每个 LLM 提供商从注册到写代码的完整流程，选一个跟着做就行。

方案 A：DeepSeek（推荐学习使用） ​

便宜、Function Calling 支持好、兼容 OpenAI SDK，最适合入门。

第一步：注册并获取 API Key

  1. 打开 https://platform.deepseek.com/
  2. 注册账号并登录
  3. 左侧菜单 → API Keys → Create API Key
  4. 复制生成的 Key（以 sk- 开头，只显示一次，务必保存）

第二步：充值

DeepSeek 需要先充值才能调用 API。左侧菜单 → 充值，最低充 10 元即可，学习阶段足够用很久。

第三步：安装依赖

bash
uv add openai

第四步：写代码

python
from openai import OpenAI

# 创建客户端——只需改 base_url 指向 DeepSeek
client = OpenAI(
    base_url="https://api.deepseek.com",       # DeepSeek 的接口地址
    api_key="sk-你的DeepSeek密钥"               # 替换为你的 API Key
)

# 调用模型
response = client.chat.completions.create(
    model="deepseek-chat",                      # DeepSeek 的模型名称
    messages=[
        {"role": "user", "content": "你好，请用一句话介绍自己"}
    ]
)

print(response.choices[0].message.content)

第五步：运行

bash
uv run python 你的文件名.py

API Key 不要硬编码

上面的示例把 Key 直接写在代码里是为了演示。实际项目中应该用环境变量：

bash
# 先设置环境变量（PowerShell）
$env:DEEPSEEK_API_KEY="sk-你的密钥"
python
import os
client = OpenAI(
    base_url="https://api.deepseek.com",
    api_key=os.environ.get("DEEPSEEK_API_KEY")
)

方案 B：OpenAI（GPT-4o） ​

效果最好，但价格较贵，需要海外信用卡。

第一步：注册并获取 API Key

  1. 打开 https://platform.openai.com/
  2. 注册账号（需要海外手机号）
  3. 右上角头像 → View API Keys → Create new secret key
  4. 复制生成的 Key（以 sk- 开头）

第二步：充值

左侧菜单 → Billing → Add payment method（需要海外信用卡），最低充 $5。

第三步：安装依赖

bash
uv add openai

第四步：写代码

python
import os
from openai import OpenAI

# OpenAI 官方——不需要改 base_url，SDK 默认就是 OpenAI 的地址
client = OpenAI(
    api_key=os.environ.get("OPENAI_API_KEY")    # 环境变量中读取 Key
)

response = client.chat.completions.create(
    model="gpt-4o-mini",                        # 便宜的学习用模型；生产用 gpt-4o
    messages=[
        {"role": "user", "content": "你好，请用一句话介绍自己"}
    ]
)

print(response.choices[0].message.content)

第五步：运行

bash
# 先设置环境变量（PowerShell）
$env:OPENAI_API_KEY="sk-你的OpenAI密钥"

# 运行
uv run python 你的文件名.py

方案 C：通义千问（Qwen） ​

国内服务，无需海外信用卡，中文效果好。

第一步：注册并获取 API Key

  1. 打开 https://dashscope.console.aliyun.com/
  2. 用阿里云账号登录
  3. 右上角头像 → API-KEY 管理 → 创建新的 API-KEY
  4. 复制生成的 Key

第二步：开通服务

左侧菜单 → 模型广场 → 找到 Qwen 系列 → 开通（有免费额度）。

第三步：安装依赖

bash
uv add openai

第四步：写代码

python
import os
from openai import OpenAI

# 通义千问——兼容 OpenAI SDK，改 base_url 即可
client = OpenAI(
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",  # 通义千问的兼容接口
    api_key=os.environ.get("DASHSCOPE_API_KEY")                    # 环境变量中读取 Key
)

response = client.chat.completions.create(
    model="qwen-plus",                          # 通义千问的模型名称
    messages=[
        {"role": "user", "content": "你好，请用一句话介绍自己"}
    ]
)

print(response.choices[0].message.content)

第五步：运行

bash
# 先设置环境变量（PowerShell）
$env:DASHSCOPE_API_KEY="sk-你的通义千问密钥"

# 运行
uv run python 你的文件名.py

方案 D：Ollama 本地模型（免费） ​

完全免费，数据不出本机，但需要显卡（或用小模型跑在 CPU 上），效果取决于模型大小。

第一步：安装 Ollama

  1. 打开 https://ollama.com/
  2. 下载 Windows 版安装包并安装
  3. 安装完成后，Ollama 会自动在后台运行

第二步：拉取模型

bash
# 推荐支持 Function Calling 的模型
ollama pull qwen2.5          # 通义千问 2.5，7B 参数，中文好

# 如果显卡显存不够（<8GB），用小模型
ollama pull qwen2.5:1.5b     # 1.5B 参数，CPU 也能跑，但效果差一些

Ollama 的 Function Calling 支持

不是所有 Ollama 模型都支持 Function Calling。目前支持较好的有：Qwen2.5、Mistral、Command-R。Llama3 系列支持有限。如果 Agent 调用工具时 LLM 不响应，换一个模型试试。

第三步：安装依赖

bash
uv add openai

第四步：写代码

python
from openai import OpenAI

# Ollama 本地——不需要 API Key，base_url 指向本地 Ollama 服务
client = OpenAI(
    base_url="http://localhost:11434/v1",        # Ollama 默认地址
    api_key="ollama"                             # 随便填，Ollama 不验证
)

response = client.chat.completions.create(
    model="qwen2.5",                             # 必须和 ollama pull 的模型名一致
    messages=[
        {"role": "user", "content": "你好，请用一句话介绍自己"}
    ]
)

print(response.choices[0].message.content)

第五步：运行

bash
# 确保 Ollama 在运行（打开浏览器访问 http://localhost:11434 看是否有响应）
uv run python 你的文件名.py

方案 E：Anthropic Claude ​

Claude 的 Function Calling 支持很好，但需要海外信用卡，且使用独立的 anthropic SDK（不兼容 OpenAI SDK）。

第一步：注册并获取 API Key

  1. 打开 https://console.anthropic.com/
  2. 注册账号
  3. Settings → API Keys → Create Key
  4. 复制生成的 Key（以 sk-ant- 开头）

第二步：充值

Plans & Billing → Add payment method（需要海外信用卡），最低充 $5。

第三步：安装依赖

bash
uv add anthropic

第四步：写代码

python
import os
import anthropic

# Claude 使用独立的 anthropic SDK，不是 openai SDK
client = anthropic.Anthropic(
    api_key=os.environ.get("ANTHROPIC_API_KEY")
)

response = client.messages.create(
    model="claude-sonnet-4-6",                   # Claude 模型名称
    max_tokens=1024,
    messages=[
        {"role": "user", "content": "你好，请用一句话介绍自己"}
    ]
)

print(response.content[0].text)

第五步：运行

bash
# 先设置环境变量（PowerShell）
$env:ANTHROPIC_API_KEY="sk-ant-你的Claude密钥"

# 运行
uv run python 你的文件名.py

::: important Claude 的 SDK 和 OpenAI 不同 Claude 用 anthropic SDK，API 格式和 openai SDK 不一样。如果你用 Claude，Agent 的 while 循环代码也要用 Anthropic 的格式（见下方第二步的 2.2 节）。其他所有方案（OpenAI / DeepSeek / 通义 / Ollama）都用同一套 openai SDK 代码。 :::

各方案速查对比 ​

DeepSeekOpenAI通义千问OllamaClaude
SDKopenaiopenaiopenaiopenaianthropic
base_urlhttps://api.deepseek.com默认（无需设置）https://dashscope.aliyuncs.com/compatible-mode/v1http://localhost:11434/v1N/A（独立 SDK）
modeldeepseek-chatgpt-4o-miniqwen-plusqwen2.5claude-sonnet-4-6
api_keysk-...sk-...sk-...随便填sk-ant-...
需要信用卡❌ 支付宝/微信✅ 海外信用卡❌ 支付宝❌ 免费✅ 海外信用卡
学习成本几毛~几元/天几元~几十元/天有免费额度免费几元~几十元/天

切换模型只需改三行

从 DeepSeek 切换到通义千问，只需要改 base_url、api_key 和 model，其余代码完全不变：

python
# DeepSeek
client = OpenAI(base_url="https://api.deepseek.com", api_key="sk-deepseek-xxx")

# 切换到通义千问——只改这三行
client = OpenAI(base_url="https://dashscope.aliyuncs.com/compatible-mode/v1", api_key="sk-dashscope-xxx")
# model 也要改：deepseek-chat → qwen-plus
bash
# 安装依赖（选一个 LLM SDK 即可）
uv add openai kubernetes          # OpenAI / DeepSeek / 通义 / Ollama 通用
uv add anthropic kubernetes       # Anthropic Claude 专用

第一步：写一个最简单的 Skill（Tool） ​

Skill 的本质就是一个 Python 函数 + 一段 JSON Schema 描述。LLM 读到 Schema 就知道什么时候该调用这个函数、该传什么参数。

1.1 写 Python 函数 ​

python
# skills/k8s.py
from kubernetes import client, config

config.load_kube_config()

def list_pods(namespace: str = "default") -> str:
    """列出指定命名空间的 Pod"""
    v1 = client.CoreV1Api()
    pods = v1.list_namespaced_pod(namespace=namespace)
    result = []
    for pod in pods.items:
        result.append(f"{pod.metadata.name} - {pod.status.phase}")
    return "\n".join(result)

这就是一个 Skill 的实现——普通 Python 函数，没什么特别的。

1.2 写 JSON Schema 描述 ​

LLM 不认识你的 Python 函数，它只认识 JSON Schema。这段描述告诉 LLM："存在这样一个工具，它接受这些参数"。

python
# skills/k8s_schema.py

list_pods_schema = {
    "type": "function",
    "function": {
        "name": "list_pods",
        "description": "列出 Kubernetes 指定命名空间中的 Pod 及其状态",
        "parameters": {
            "type": "object",
            "properties": {
                "namespace": {
                    "type": "string",
                    "description": "Kubernetes 命名空间，默认为 default"
                }
            },
            "required": []
        }
    }
}

关键点

  • name 必须和 Python 函数名一致
  • description 非常重要——LLM 靠它判断"什么时候该调用这个工具"
  • 每个参数的 description 也要写清楚，LLM 靠它决定传什么值

1.3 把函数和 Schema 绑在一起 ​

python
# skills/registry.py

# 工具注册表：Schema 描述 → Python 函数
TOOL_REGISTRY = {
    "list_pods": list_pods,  # 函数引用
}

TOOLS_SCHEMAS = [list_pods_schema]  # 传给 LLM 的描述列表

至此，一个完整的 Skill 就写好了：函数（实现）+ Schema（描述）+ 注册（绑定）。

第二步：写一个最简单的 Agent ​

Agent 的本质就是一个 while 循环：把用户消息发给 LLM → LLM 说要调用工具 → 你执行工具 → 把结果喂回 LLM → 重复。

2.1 用 OpenAI SDK 实现 ​

python
# agent.py
import json
from openai import OpenAI
from skills.registry import TOOL_REGISTRY, TOOLS_SCHEMAS

client = OpenAI()

def run_agent(user_message: str):
    messages = [
        {
            "role": "system",
            "content": "你是一个 Kubernetes 运维助手。用户会用自然语言描述需求，你通过调用工具来操作集群。"
        },
        {
            "role": "user",
            "content": user_message
        }
    ]

    while True:
        # 1. 把对话发给 LLM
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=TOOLS_SCHEMAS  # 告诉 LLM 有哪些工具可用
        )

        message = response.choices[0].message

        # 2. 如果 LLM 不想调用工具，说明它已经完成了，直接返回
        if not message.tool_calls:
            print("Agent:", message.content)
            break

        # 3. 把 LLM 的回复加入对话历史
        messages.append(message)

        # 4. 执行 LLM 请求的工具调用
        for tool_call in message.tool_calls:
            function_name = tool_call.function.name
            function_args = json.loads(tool_call.function.arguments)

            print(f"调用工具: {function_name}({function_args})")

            # 从注册表找到对应函数并执行
            result = TOOL_REGISTRY[function_name](**function_args)

            # 5. 把工具执行结果喂回 LLM
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": str(result)
            })

        # 继续循环，让 LLM 看到工具结果后决定下一步

2.2 用 Anthropic SDK 实现 ​

python
# agent_anthropic.py
import json
import anthropic
from skills.registry import TOOL_REGISTRY, TOOLS_SCHEMAS

client = anthropic.Anthropic()

def run_agent(user_message: str):
    messages = [{"role": "user", "content": user_message}]

    while True:
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            system="你是一个 Kubernetes 运维助手。用户会用自然语言描述需求，你通过调用工具来操作集群。",
            messages=messages,
            tools=TOOLS_SCHEMAS
        )

        # 检查是否有工具调用
        tool_use_blocks = [b for b in response.content if b.type == "tool_use"]
        text_blocks = [b for b in response.content if b.type == "text"]

        # 打印文本回复
        for block in text_blocks:
            if block.text:
                print("Agent:", block.text)

        # 没有工具调用，结束
        if not tool_use_blocks:
            break

        # 执行工具调用
        tool_results = []
        for block in tool_use_blocks:
            print(f"调用工具: {block.name}({block.input})")
            result = TOOL_REGISTRY[block.name](**block.input)
            tool_results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": str(result)
            })

        # 把结果喂回 LLM
        messages.append({"role": "assistant", "content": response.content})
        messages.append({"role": "user", "content": tool_results})

2.3 运行 ​

详细的运行环境准备、API Key 配置和常见问题，请参考 运行指南。

python
run_agent("帮我看看 default 命名空间有哪些 Pod")

你会看到类似这样的执行过程：

调用工具: list_pods({"namespace": "default"})
Agent: default 命名空间中当前有以下 Pod：
- nginx-deployment-abc123 - Running
- redis-master-0 - Running
- my-app-xyz789 - Pending

这就是一个完整的 Agent！核心逻辑就是一个 while 循环 + 函数分发。

第三步：添加更多 Skill ​

现在你已经有了基本骨架，添加新 Skill 只需要三步：

3.1 写函数 ​

python
# skills/k8s.py（续）

def get_pod_logs(name: str, namespace: str = "default", tail_lines: int = 50) -> str:
    """获取 Pod 日志"""
    v1 = client.CoreV1Api()
    logs = v1.read_namespaced_pod_log(
        name=name, namespace=namespace, tail_lines=tail_lines
    )
    return logs

def list_deployments(namespace: str = "default") -> str:
    """列出 Deployment"""
    apps_v1 = client.AppsV1Api()
    deploys = apps_v1.list_namespaced_deployment(namespace=namespace)
    result = []
    for d in deploys.items:
        ready = d.status.ready_replicas or 0
        result.append(f"{d.metadata.name} - {ready}/{d.spec.replicas}")
    return "\n".join(result)

3.2 写 Schema ​

python
get_pod_logs_schema = {
    "type": "function",
    "function": {
        "name": "get_pod_logs",
        "description": "获取指定 Kubernetes Pod 的日志",
        "parameters": {
            "type": "object",
            "properties": {
                "name": {
                    "type": "string",
                    "description": "Pod 名称"
                },
                "namespace": {
                    "type": "string",
                    "description": "命名空间，默认 default"
                },
                "tail_lines": {
                    "type": "integer",
                    "description": "返回的最后 N 行日志，默认 50"
                }
            },
            "required": ["name"]
        }
    }
}

list_deployments_schema = {
    "type": "function",
    "function": {
        "name": "list_deployments",
        "description": "列出指定命名空间中的 Deployment 及其就绪状态",
        "parameters": {
            "type": "object",
            "properties": {
                "namespace": {
                    "type": "string",
                    "description": "命名空间，默认 default"
                }
            },
            "required": []
        }
    }
}

3.3 注册 ​

python
# skills/registry.py（更新）

TOOL_REGISTRY = {
    "list_pods": list_pods,
    "get_pod_logs": get_pod_logs,
    "list_deployments": list_deployments,
}

TOOLS_SCHEMAS = [
    list_pods_schema,
    get_pod_logs_schema,
    list_deployments_schema,
]

Agent 代码不需要改任何东西——它会自动识别新增的工具。

第四步：用框架简化开发 ​

手写 while 循环能帮你理解原理，但实际开发中用框架可以省去大量重复代码。

4.1 用 LangChain 快速实现 ​

python
# agent_langchain.py
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate

# 用装饰器定义 Tool（函数 + Schema 一步到位）
@tool
def list_pods(namespace: str = "default") -> str:
    """列出 Kubernetes 指定命名空间中的 Pod 及其状态"""
    v1 = client.CoreV1Api()
    pods = v1.list_namespaced_pod(namespace=namespace)
    return "\n".join(f"{p.metadata.name} - {p.status.phase}" for p in pods.items)

@tool
def get_pod_logs(name: str, namespace: str = "default") -> str:
    """获取指定 Kubernetes Pod 的日志"""
    v1 = client.CoreV1Api()
    return v1.read_namespaced_pod_log(name=name, namespace=namespace)

@tool
def list_deployments(namespace: str = "default") -> str:
    """列出指定命名空间中的 Deployment 及其就绪状态"""
    apps_v1 = client.AppsV1Api()
    deploys = apps_v1.list_namespaced_deployment(namespace=namespace)
    return "\n".join(
        f"{d.metadata.name} - {d.status.ready_replicas or 0}/{d.spec.replicas}"
        for d in deploys.items
    )

# 创建 Agent
llm = ChatOpenAI(model="gpt-4o")
tools = [list_pods, get_pod_logs, list_deployments]

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个 Kubernetes 运维助手。通过调用工具来操作集群。"),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

agent = create_tool_calling_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 运行
result = agent_executor.invoke({"input": "default 命名空间有哪些 Pod？"})
print(result["output"])

对比

  • 手写：~60 行，理解原理，灵活控制
  • LangChain：~30 行，@tool 装饰器把函数和 Schema 合一，Agent 循环自动处理
  • 选择建议：先手写一遍理解原理，再用框架提高效率

4.2 用 MCP 做可复用的 Skill 服务 ​

如果你希望 Skill 能被不同的 Agent、不同的 LLM 调用，用 MCP 把 Skill 封装成独立服务：

python
# mcp_k8s_server.py
from mcp.server.fastmcp import FastMCP
from kubernetes import client, config

config.load_kube_config()
mcp = FastMCP("k8s-manager")

@mcp.tool()
def list_pods(namespace: str = "default") -> str:
    """列出指定命名空间的 Pod"""
    v1 = client.CoreV1Api()
    pods = v1.list_namespaced_pod(namespace=namespace)
    return "\n".join(f"{p.metadata.name} - {p.status.phase}" for p in pods.items)

@mcp.tool()
def get_pod_logs(name: str, namespace: str = "default") -> str:
    """获取 Pod 日志"""
    v1 = client.CoreV1Api()
    return v1.read_namespaced_pod_log(name=name, namespace=namespace)

@mcp.tool()
def list_deployments(namespace: str = "default") -> str:
    """列出 Deployment"""
    apps_v1 = client.AppsV1Api()
    deploys = apps_v1.list_namespaced_deployment(namespace=namespace)
    return "\n".join(
        f"{d.metadata.name} - {d.status.ready_replicas or 0}/{d.spec.replicas}"
        for d in deploys.items
    )

mcp.run()

这样任何支持 MCP 的客户端（Claude Desktop、Cursor、其他 Agent 框架）都能直接调用你的 K8s Skill。

总结：构建路线图 ​

理解概念                    动手实践                     进阶提升
─────────────────────────────────────────────────────────────────
LLM API ──────→  写 Python 函数 ──────→  用框架（LangChain / CrewAI）
Function Calling ──→  写 JSON Schema ──────→  用 MCP 做可复用服务
JSON Schema ──────→  写 while 循环 ──────→  多 Agent 协作
System Prompt ────→  跑通第一个 Agent ──────→  加 Memory / RAG
ReAct 循环 ───────→  逐步添加 Skill ──────→  生产级部署

最快上手路径：装好 SDK → 写一个 Python 函数 → 写对应的 JSON Schema → 手写 while 循环跑通 → 再考虑用框架。

参考资料 ​

基于 MIT 许可发布

版权所有 © 2024-present Walter