Claude & AI Agent 实战

Claude & AI Agent 开发实战指南

从零基础到全栈 AI 工程师的完整路径 — 大模型原理、Token 机制、API 调用、Prompt Engineering、AI Agent 架构、RAG 系统、MCP 协议到生产级项目落地,一本手册搞定。

零基础友好 Claude API & SDK Claude Code CLI AI Agent 开发 MCP 协议 RAG 系统 LangChain / LlamaIndex Multi-Agent
0

AI 大模型基础入门

零基础必读 — 大模型原理、Token、参数量、提供商对比、选型指南
写在前面本章面向零基础读者,帮助你建立 AI 开发的核心认知地图。如果你已经了解 LLM 基础,可以直接跳到第一章。但建议先快速浏览 0.5 节(价格对比),很多老手也会忽略。

0.1 什么是大语言模型(LLM)

从"查字典"到"理解语言" — AI 的进化史

在大模型出现之前,计算机处理语言的方式非常机械:程序员要手工编写规则("如果用户说'你好',就回答'您好'"),或者用统计方法统计词语出现频率。这些方法在面对真实世界的复杂语言时,很快就捉襟见肘。

🗂 规则系统(1950s-1990s)

手工编写语法规则,面对真实语言的多样性完全不够用。

📊 统计模型(1990s-2010s)

n-gram、HMM,通过统计词语共现概率预测下一个词。

🧠 神经网络(2013-2017)

Word2Vec、RNN/LSTM,让词语有了"语义距离"的概念。

⚡ Transformer + 预训练(2017至今)

GPT、BERT 开创了"大规模预训练 + 微调"的范式,LLM 时代开始。

大语言模型的本质:一个极其复杂的"下一个词预测器"

听起来很简单,但正是这个简单任务,催生了令人惊叹的智能。LLM 的训练目标就是:给定前面所有的词,预测下一个词的概率分布

今天 天气 真的 LLM 神经网络 热 45% 好 30% 棒 15% 输入 Token 序列 下一词概率

为了能准确预测下一个词,模型必须理解语法、语义、常识、逻辑、甚至情感。这就是为什么一个单纯预测词语的模型,最终展现出了"智能"的能力。

Transformer 架构:LLM 的心脏

2017 年 Google 发表了论文《Attention Is All You Need》,提出了 Transformer 架构,成为了几乎所有现代 LLM 的基础。它最核心的创新是自注意力机制(Self-Attention)

自注意力的直觉理解处理"银行"这个词时,机器如何知道它是"河岸"还是"金融机构"?自注意力让模型计算这个词与句子中所有其他词的相关程度——如果上下文有"存款""贷款",注意力权重就高度集中在金融语义上。

预训练与微调:LLM 学习的两个阶段

📚 预训练(Pre-training)

在海量文本(互联网、书籍、代码)上学习语言规律,形成通用能力。代价:数千万美元的算力。

🎯 微调(Fine-tuning / RLHF)

在特定任务数据或人类反馈上继续训练,让模型"听话"、"安全"、"专业"。ChatGPT 背后的关键。

🧩 提示词工程(Prompting)

不改变模型权重,仅通过输入文本引导模型行为。成本极低,是开发者最常用的方式。

0.2 Token — AI 的基本计量单位

什么是 Token?

LLM 不是逐字处理文字,而是将文本拆分为 Token(词元)再处理。Token 是模型能识别的最小语言单元,既不完全等于"词",也不等于"字符"。

Token 的直观感受(英文) "Hello world" → ["Hello", " world"] = 2 tokens
"ChatGPT is amazing" → ["Chat", "G", "PT", " is", " amazing"] = 5 tokens
通常 1 token ≈ 0.75 个英文单词,或 100 tokens ≈ 75 个单词。
中文 Token 的特殊性 中文每个汉字通常是 1-2 个 token(视分词器而定)。这意味着同样的内容,中文消耗的 token 数通常比英文多 1.5-2 倍,直接影响 API 成本。

"你好世界" → 4 tokens(约)
"Hello World" → 2 tokens
所以用英文 prompt 会更省钱,但要权衡准确度。

Token 的重要性

💰 直接决定成本

API 按 token 计费。每次请求 = 输入 token 数 × 输入单价 + 输出 token 数 × 输出单价。

📏 决定上下文长度

模型有最大 token 限制(上下文窗口)。超出则需截断,可能丢失关键信息。

⏱ 影响速度

输出 token 越多,响应越慢。生成 1000 tokens 比生成 100 tokens 慢约 10 倍。

如何估算 Token 数

大部分 AI 提供商提供 Tokenizer 工具,开发时可用于精确计算。下面是快速估算规则:

语言/内容大致换算举例
英文1 token ≈ 4 个字符 / 0.75 个单词"artificial intelligence" = 3 tokens
中文1 个汉字 ≈ 1.5 tokens"人工智能" ≈ 6 tokens
代码1 token ≈ 3-4 个字符100 行代码 ≈ 600-800 tokens
图片(Vision)按分辨率计算1024×1024 图 ≈ 765-1500 tokens

上下文窗口(Context Window)

上下文窗口是模型在一次请求中能"看到"的最大 token 数,包括输入 + 输出。超过这个限制,模型会"忘记"更早的内容。

系统提示词 历史对话记录 当前用户输入 模型输出空间 剩余 上下文窗口(例:200K tokens) 当历史对话超出窗口时,最早的内容会被截断,模型将"遗忘"
模型上下文窗口约等于
Claude 系列(Opus/Sonnet/Haiku)200K tokens约 15 万字 / 500 页书
GPT-4o128K tokens约 10 万字 / 330 页书
Gemini 1.5 Pro1M tokens约 75 万字 / 2500 页书
Gemini 2.0 Flash1M tokens约 75 万字 / 2500 页书

0.3 模型参数量 — 理解模型规模

参数是什么?

神经网络本质上是一个巨大的数学函数,由数十亿个可调节的数值(参数/权重)构成。训练过程就是不断调整这些参数,让模型的预测越来越准确。

直观类比把参数想象成一张巨大的"知识网络"中的连接强度。7B 参数的模型有 70 亿个这样的连接,每个连接都记录了从海量文本中学到的某种语言模式或知识。参数越多,能记住的"知识"就越多,理解能力通常也越强。

主流模型参数规模参考

参数量代表模型能力描述部署需求
1B-7BLlama-3.2-3B, Phi-3-mini基础对话,简单任务,速度快普通笔记本可运行
7B-13BLlama-3.1-8B, Mistral-7B日常任务良好,有一定推理能力8GB 显存(消费级 GPU)
30B-70BLlama-3.3-70B, Qwen-2.5-72B复杂推理,接近 GPT-4 水平2-4 块高端 GPU
100B+GPT-4(未公开), Claude(未公开)顶级能力,专业任务表现卓越大规模 GPU 集群
参数量 ≠ 一切现代训练技术让"小参数量"模型也可以极其强大。例如 Google 的 Gemma-2-9B 在某些基准上超越了 70B 的旧模型。训练数据质量、训练方法、RLHF 对齐对最终效果的影响,往往大于单纯的参数量。

量化(Quantization):缩小模型大小的技巧

原始模型的每个参数使用 32 位浮点数(FP32)存储。量化将参数压缩到 8 位(INT8)甚至 4 位(INT4),使模型体积缩小 4-8 倍,只有轻微的性能损失。

精度内存占用(以 7B 模型为例)性能损失适用场景
FP32(全精度)~28 GB训练
FP16 / BF16~14 GB极少推理(服务器)
INT8~7 GB很少推理(高端消费 GPU)
INT4 / Q4~3.5 GB轻微本地运行(普通电脑)

0.4 核心概念速查

Temperature — 控制输出的"创意度"

Temperature 是控制模型输出随机性的参数,范围通常为 0.0 到 1.0(部分模型到 2.0)。

行为适用场景
0.0完全确定性,每次输出相同代码生成、数据提取、分类任务
0.3-0.5轻微随机,以准确为主翻译、总结、问答
0.7-0.9适度创意写作助手、头脑风暴、对话机器人
1.0+高随机性、发散创意写作、诗歌、多样化生成

Embedding(向量嵌入) — 把文字变成数字

Embedding 是将文字转换为数字向量的技术,使计算机能够计算文本之间的"语义距离"。这是 RAG(检索增强生成)系统的核心基础。

直观理解"猫"和"狗"的向量在数学空间上很近;"猫"和"飞机"的向量很远。通过余弦相似度计算,可以找到语义最相近的文本片段。

RLHF — 让 AI 变得"有用且安全"

Reinforcement Learning from Human Feedback(人类反馈强化学习)是将"好"与"坏"的人类判断注入模型训练的方法,是 ChatGPT、Claude 遵守指令、拒绝有害内容的关键技术。

1️⃣ 监督微调(SFT)

用高质量的人工撰写示例对模型进行微调,教会模型"怎么回答"。

2️⃣ 奖励模型(RM)

人类对模型的多个回答打分排名,训练一个"评委"模型。

3️⃣ PPO 强化学习

用奖励模型作为"反馈信号",不断优化主模型,使其获得更高分数。

RAG(检索增强生成)— 给 AI 接上"外部记忆"

LLM 的训练数据有截止日期,且无法记住公司内部文档。RAG 通过在回答前先从外部数据库检索相关内容,再让 LLM 基于检索结果生成回答,解决了这两个问题。

类比RAG 就像开卷考试——学生(LLM)不需要背住所有知识,可以在回答时查阅资料(向量数据库),再综合作答。第五章将深入介绍 RAG 实现。

Hallucination(幻觉)— LLM 最大的缺陷

LLM 有时会自信地生成看起来合理但实际错误的内容,称为"幻觉"。因为模型只是在做概率预测,不具备真正的知识检索能力。

类型示例应对方法
事实幻觉捏造不存在的引用文献RAG、要求模型引用来源
推理错误复杂数学题计算错误Chain-of-Thought、代码执行
指令遗忘长对话后忘记系统要求定期重申 system prompt

0.5 主流 AI 提供商全景 & 价格对比 必看

主要 AI 提供商与旗舰模型(2025-2026)

🟣 Anthropic — Claude 系列

以安全性和指令遵循见长,长文本处理出色,200K 上下文。编程和分析任务表现顶级。

🟢 OpenAI — GPT 系列

开创 ChatGPT 时代,生态最成熟,插件/工具调用最丰富,行业事实标准。

🔵 Google — Gemini 系列

超长上下文(1M tokens),原生多模态,与 Google 生态深度集成。

🟡 Meta — Llama 系列

开源免费,可本地部署,避免数据隐私问题,企业私有化部署首选。

⚪ Mistral — 开源高效模型

欧洲团队,开源旗舰性价比极高,Mistral Large 接近 GPT-4 水平。

🔴 阿里云 — Qwen 系列

中文能力最强,本土化优秀,开源版本可免费使用,商业版价格极低。

🟠 DeepSeek — R1/V3 系列

数学和代码能力突出,开源且成本极低,中国团队,近期表现震惊业界。

🟤 xAI — Grok 系列

马斯克旗下,实时联网(X平台数据),Grok-3 竞争顶级梯队。

2025-2026 主流模型 API 定价对比

注意以下价格均为每 1M tokens(百万 token)的美元价格,每家计费方式略有差异。价格会随竞争持续下降,请以官网为准。
提供商 模型 输入价格 $/1M tokens 输出价格 $/1M tokens 上下文 适用场景
AnthropicClaude Opus 4.6$15$75200K最强推理/复杂任务
AnthropicClaude Sonnet 4.6$3$15200K均衡推荐,日常首选
AnthropicClaude Haiku 4.5$0.8$4200K高频轻量任务
OpenAIGPT-4o$2.5$10128K通用旗舰
OpenAIGPT-4o mini$0.15$0.6128K低成本高频
OpenAIo1$15$60200K高级推理(慢)
GoogleGemini 2.0 Flash$0.1$0.41M超低成本、超长文
GoogleGemini 1.5 Pro$1.25$52M超长文档分析
DeepSeekDeepSeek-V3$0.27$1.164K极低成本通用
DeepSeekDeepSeek-R1$0.55$2.1964K数学/代码推理
阿里云Qwen-Max¥0.04/千tokens¥0.12/千tokens32K中文任务首选
阿里云Qwen-Long¥0.0005/千tokens¥0.002/千tokens10M超长文档/极低成本

成本计算实例

假设你要构建一个每日处理 10,000 次用户请求 的客服机器人,每次平均 输入 500 tokens + 输出 200 tokens

模型每日成本(美元)每月成本(美元)备注
Claude Opus 4.6$18.75~$563最强效果,成本较高
Claude Sonnet 4.6$4.5~$135推荐首选
Claude Haiku 4.5$1.2~$36轻量任务最省钱
GPT-4o mini$0.195~$5.85极低成本,能力有限
Gemini 2.0 Flash$0.13~$3.9当前最便宜旗舰级
DeepSeek-V3$0.355~$10.65国产高性价比
省钱小技巧
  • Prompt Caching:对重复的 system prompt 使用缓存,可节省 80-90% 的输入成本(Claude 支持)
  • 模型分级:简单问题用 Haiku,复杂问题再升级到 Sonnet/Opus
  • 输出长度控制:明确要求"简洁回答",输出 token 通常比输入贵 3-5 倍
  • 批量 API:非实时任务用 Batch API,价格是实时的 50%

0.6 如何选择适合的模型

决策树:选模型不再迷茫

开始选模型 数据不能上云(隐私/合规)? 本地部署 Llama 3 / Qwen 成本极度敏感? 低成本方案 Haiku / GPT-4o mini Gemini Flash / DeepSeek 任务类型? 代码/数学 通用 中文 Claude Sonnet / DeepSeek-R1 Claude Sonnet / GPT-4o Qwen-Max / Claude+中文prompt 需要最强效果? 否 → 先用 Sonnet Claude Opus 4.6

按使用场景推荐

场景推荐模型理由
学习/个人项目Claude Sonnet 4.6 / GPT-4o能力强,有免费额度,生态好
高并发客服机器人Claude Haiku 4.5 / GPT-4o mini成本低,速度快,足够处理简单问题
代码辅助/编程AgentClaude Sonnet 4.6 / DeepSeek-R1代码能力顶尖
长文档分析(100页+)Gemini 1.5 Pro / Claude Sonnet超长上下文窗口
中文内容生成Qwen-Max / Claude + 中文 prompt中文语感更自然
数据隐私/私有化部署Llama 3.3-70B / Qwen-2.5-72B开源,可本地运行
复杂科学推理Claude Opus 4.6 / o1最强推理链
图片/多模态理解GPT-4o / Claude Sonnet(Vision)多模态能力成熟
入门建议如果你刚开始学习 AI 开发,Claude Sonnet 4.6 是最推荐的起点:能力强、价格合理、200K 长上下文、指令遵循好,本文档后续所有示例均以 Claude 系列为主。
1

Claude API 完全指南

Anthropic SDK、对话、工具调用、多模态、流式输出

1.1 API 快速入门 API

Claude 模型家族概览(2025-2026)

模型Model ID定位上下文窗口特点
Claude Opus 4.6claude-opus-4-6最强旗舰200K最强推理与编程能力
Claude Sonnet 4.6claude-sonnet-4-6性能均衡200K速度与智能的最佳平衡
Claude Haiku 4.5claude-haiku-4-5-20251001快速轻量200K低延迟、低成本
Python
# === 安装 Anthropic Python SDK ===
# pip install anthropic

import anthropic

# 初始化客户端(API Key 从环境变量 ANTHROPIC_API_KEY 自动读取)
client = anthropic.Anthropic()

# === 最简单的 API 调用 ===
message = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[
        {"role": "user", "content": "用一句话解释什么是Transformer架构?"}
    ]
)
print(message.content[0].text)
# 输出: "Transformer是一种基于自注意力机制的神经网络架构..."

# === 请求结构详解 ===
message = client.messages.create(
    model="claude-sonnet-4-6",  # 模型选择
    max_tokens=2048,             # 最大生成token数
    temperature=0.7,             # 0=确定性, 1=更随机创意
    top_p=0.9,                   # 核采样(与temperature配合)
    system="你是一位资深Python架构师,回答简洁专业。",  # 系统提示
    messages=[                   # 对话消息列表
        {"role": "user", "content": "FastAPI和Flask的区别?"}
    ],
    stop_sequences=["\n\n---"],  # 遇到这个序列就停止生成
)

# === 响应结构 ===
print(f"模型: {message.model}")
print(f"停止原因: {message.stop_reason}")  # end_turn / max_tokens / stop_sequence
print(f"输入Token: {message.usage.input_tokens}")
print(f"输出Token: {message.usage.output_tokens}")
print(f"内容: {message.content[0].text}")
TypeScript
// === Node.js / TypeScript SDK ===
// npm install @anthropic-ai/sdk

import Anthropic from "@anthropic-ai/sdk";

const client = new Anthropic();

async function main() {
  const message = await client.messages.create({
    model: "claude-sonnet-4-6",
    max_tokens: 1024,
    system: "你是一位全栈工程师。",
    messages: [
      { role: "user", content: "如何设计一个高可用的微服务架构?" }
    ],
  });

  console.log(message.content[0].type === "text"
    ? message.content[0].text
    : "非文本响应");
}

main();
cURL
# === REST API 直接调用 ===
curl https://api.anthropic.com/v1/messages \
  -H "Content-Type: application/json" \
  -H "x-api-key: $ANTHROPIC_API_KEY" \
  -H "anthropic-version: 2023-06-01" \
  -d '{
    "model": "claude-sonnet-4-6",
    "max_tokens": 1024,
    "messages": [
      {"role": "user", "content": "Hello, Claude!"}
    ]
  }'

1.2 多轮对话 API

对话状态管理

Claude API 是无状态的——每次请求都需要带上完整的对话历史。这和有状态的 WebSocket 聊天不同,你需要自己管理 messages 列表。

Python
import anthropic

client = anthropic.Anthropic()

class ChatSession:
    """管理多轮对话的会话类"""

    def __init__(self, model="claude-sonnet-4-6", system_prompt=""):
        self.model = model
        self.system = system_prompt
        self.messages = []       # 对话历史
        self.total_tokens = 0    # token 用量追踪

    def send(self, user_message: str) -> str:
        """发送消息并获取回复"""
        self.messages.append({"role": "user", "content": user_message})

        response = client.messages.create(
            model=self.model,
            max_tokens=2048,
            system=self.system,
            messages=self.messages,
        )

        assistant_text = response.content[0].text
        self.messages.append({"role": "assistant", "content": assistant_text})
        self.total_tokens += response.usage.input_tokens + response.usage.output_tokens

        return assistant_text

    def clear(self):
        """清空对话历史(开始新话题时使用)"""
        self.messages = []

    def get_context_length(self):
        """估算当前上下文长度"""
        return sum(len(m["content"]) for m in self.messages) // 4  # 粗略估算

# === 使用示例 ===
chat = ChatSession(
    system_prompt="你是一位数据库专家,擅长SQL优化和架构设计。请用中文回答。"
)

# 第一轮
reply1 = chat.send("如何优化一个慢查询?")
print(f"[Claude]: {reply1}")

# 第二轮(Claude记得之前的对话)
reply2 = chat.send("能给一个具体的索引优化例子吗?")
print(f"[Claude]: {reply2}")

# 第三轮(继续深入)
reply3 = chat.send("如果是联合索引呢?最左前缀原则是什么?")
print(f"[Claude]: {reply3}")

print(f"\n总Token消耗: {chat.total_tokens}")
上下文窗口管理策略

Claude 支持 200K token 上下文,但更长的上下文意味着更高的成本和延迟。实践中常用以下策略:

  • 滑动窗口:只保留最近N轮对话
  • 摘要压缩:用 Claude 将长对话摘要成短文本
  • 分层记忆:重要信息放system prompt,次要放messages

1.3 Tool Use / Function Calling 核心

让 Claude 调用你的代码

Tool Use 是构建 AI Agent 的基础能力。你定义工具的名称、描述和参数,Claude 判断何时需要使用工具,并生成调用参数。你的代码执行工具后,将结果返回给 Claude 继续处理。

Claude 如何判断是否需要工具?

Claude 并没有什么特殊的"判断逻辑代码", 它就是通过训练,学会了理解 "我能做什么 / 我不能做什么"

你发送的 Prompt 实际上包含:
┌─────────────────────────────────────┐
│  System Prompt(角色设定)            │
│  + Tools 定义(工具列表)             │
│  + 用户消息                          │
└─────────────────────────────────────┘
// Tools 定义本质上就是一段 "特殊格式的文字",Claude 读懂了这段文字,才知道有哪些工具可用
          ↓
     全部变成 Token
          ↓
     Claude 预测下一个输出
Claude 内部的"推理"过程:
 
示例1:
Claude 看到:
  1. 用户问:今天天气怎么样
  2. 我有个工具叫 get_weather
    描述是:获取实时天气信息
Claude 推理:
  - 天气是实时数据 → 我的训练数据没有 ✗
  - get_weather 工具描述匹配这个需求 ✓
  - 结论:调用工具

示例2:
Claude 看到:
  1. 用户问:什么是递归
  2. 我有个工具叫 get_weather
Claude 推理:
  - 递归是编程概念 → 我知道 ✓
  - get_weather 和这个问题不相关 ✗
  - 结论:直接回答
Python
import anthropic
import json

client = anthropic.Anthropic()

# === 定义工具 ===
tools = [
    {
        "name": "get_weather",
        "description": "获取指定城市的当前天气信息。当用户询问天气时使用此工具。",
        "input_schema": {
            "type": "object",
            "properties": {
                "city": {
                    "type": "string",
                    "description": "城市名称,如 '北京'、'上海'"
                },
                "unit": {
                    "type": "string",
                    "enum": ["celsius", "fahrenheit"],
                    "description": "温度单位,默认celsius"
                }
            },
            "required": ["city"]
        }
    },
    {
        "name": "search_database",
        "description": "在产品数据库中搜索商品信息。",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "搜索关键词"},
                "category": {
                    "type": "string",
                    "enum": ["electronics", "clothing", "food"],
                    "description": "商品类别"
                },
                "max_results": {
                    "type": "integer",
                    "description": "最大返回数量,默认5",
                    "default": 5
                }
            },
            "required": ["query"]
        }
    },
    {
        "name": "execute_python",
        "description": "执行Python代码并返回结果。用于数学计算、数据处理等。",
        "input_schema": {
            "type": "object",
            "properties": {
                "code": {"type": "string", "description": "要执行的Python代码"}
            },
            "required": ["code"]
        }
    }
]

# === 工具执行函数 ===
def execute_tool(tool_name: str, tool_input: dict) -> str:
    """根据工具名称执行对应的操作"""
    if tool_name == "get_weather":
        # 实际项目中调用天气API
        return json.dumps({
            "city": tool_input["city"],
            "temperature": 22,
            "condition": "晴",
            "humidity": "45%",
            "wind": "东北风3级"
        }, ensure_ascii=False)

    elif tool_name == "search_database":
        # 实际项目中查询数据库
        return json.dumps({
            "results": [
                {"name": "iPhone 15 Pro", "price": 7999, "stock": 128},
                {"name": "iPhone 15", "price": 5999, "stock": 256},
            ],
            "total": 2
        }, ensure_ascii=False)

    elif tool_name == "execute_python":
        try:
            # 警告:生产环境务必用沙箱执行!
            result = {}
            exec(tool_input["code"], {"__builtins__": __builtins__}, result)
            return str(result.get("output", "执行完成"))
        except Exception as e:
            return f"执行错误: {str(e)}"

    return "未知工具"

# === 完整的 Tool Use 循环 ===
def chat_with_tools(user_message: str) -> str:
    """支持工具调用的完整对话函数"""
    messages = [{"role": "user", "content": user_message}]

    while True:
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=4096,
            tools=tools,
            messages=messages,
        )

        # 检查是否需要调用工具
        if response.stop_reason == "tool_use":
            # 收集所有工具调用
            assistant_content = response.content
            tool_results = []

            for block in assistant_content:
                if block.type == "tool_use":
                    print(f"  [调用工具] {block.name}({json.dumps(block.input, ensure_ascii=False)})")
                    result = execute_tool(block.name, block.input)
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": result,
                    })

            # 将工具调用和结果都加入对话历史
            messages.append({"role": "assistant", "content": assistant_content})
            messages.append({"role": "user", "content": tool_results})

        else:
            # 没有工具调用,返回最终文本
            return response.content[0].text

# === 测试 ===
# result = chat_with_tools("北京今天天气怎么样?适合穿什么衣服?")
# print(result)
# Claude会先调用get_weather,然后基于天气数据给出穿衣建议

# result = chat_with_tools("帮我算一下 fibonacci(20) 的值")
# print(result)
# Claude会调用execute_python来计算
Tool Use 最佳实践
  • 工具描述要清晰具体——Claude 根据描述决定是否调用
  • 参数的 description 字段很重要,帮助 Claude 生成正确参数
  • 控制工具数量(通常不超过20个),过多工具会降低选择准确率
  • 使用 tool_choice 参数可以强制使用或禁止使用特定工具

1.4 Vision 多模态 多模态

Python
import anthropic
import base64
import httpx

client = anthropic.Anthropic()

# === 方式1:URL 图片 ===
message = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[{
        "role": "user",
        "content": [
            {
                "type": "image",
                "source": {
                    "type": "url",
                    "url": "https://example.com/chart.png"
                }
            },
            {
                "type": "text",
                "text": "请分析这张图表的数据趋势和关键发现。"
            }
        ]
    }]
)

# === 方式2:Base64 编码(本地图片) ===
def analyze_local_image(image_path: str, question: str) -> str:
    with open(image_path, "rb") as f:
        image_data = base64.standard_b64encode(f.read()).decode("utf-8")

    # 根据文件扩展名判断类型
    media_type = "image/png" if image_path.endswith(".png") else "image/jpeg"

    message = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=2048,
        messages=[{
            "role": "user",
            "content": [
                {
                    "type": "image",
                    "source": {
                        "type": "base64",
                        "media_type": media_type,
                        "data": image_data,
                    }
                },
                {"type": "text", "text": question}
            ]
        }]
    )
    return message.content[0].text

# === 实际应用场景 ===
# 1. 代码截图 → 代码文本(OCR + 理解)
# result = analyze_local_image("screenshot.png", "请将截图中的代码提取出来,修复其中的bug")

# 2. 架构图分析
# result = analyze_local_image("arch.png", "分析这个系统架构图,指出潜在的单点故障")

# 3. UI设计稿 → 前端代码
# result = analyze_local_image("design.png", "根据这个设计稿,用React + TailwindCSS实现")

# 4. 多张图片对比
# message = client.messages.create(
#     model="claude-sonnet-4-6",
#     max_tokens=2048,
#     messages=[{
#         "role": "user",
#         "content": [
#             {"type": "image", "source": {"type": "base64", ...}},  # 图1
#             {"type": "image", "source": {"type": "base64", ...}},  # 图2
#             {"type": "text", "text": "对比这两个版本的UI,列出主要差异"}
#         ]
#     }]
# )

1.5 Streaming 流式输出 实战

Python
import anthropic

client = anthropic.Anthropic()

# === 流式输出 — 提升用户体验的关键 ===
# 不需要等完整响应生成完才显示,实时逐字输出

# 方式1:with 语法(推荐)
with client.messages.stream(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[{"role": "user", "content": "写一首关于编程的诗"}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

# 获取最终的完整消息
final_message = stream.get_final_message()
print(f"\n\n总Token: {final_message.usage.input_tokens + final_message.usage.output_tokens}")

# === 方式2:事件处理(更精细的控制) ===
with client.messages.stream(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello"}],
) as stream:
    for event in stream:
        if event.type == "content_block_delta":
            if event.delta.type == "text_delta":
                print(event.delta.text, end="")
        elif event.type == "message_stop":
            print("\n[生成完成]")

# === 在 Web 应用中使用 SSE 流式传输 ===
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

@app.get("/chat/stream")
async def stream_chat(question: str):
    def generate():
        with client.messages.stream(
            model="claude-sonnet-4-6",
            max_tokens=2048,
            messages=[{"role": "user", "content": question}],
        ) as stream:
            for text in stream.text_stream:
                yield f"data: {text}\n\n"  # SSE格式
        yield "data: [DONE]\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")

1.6 Prompt Caching — 降低成本的利器 节省90%

Python
import anthropic

client = anthropic.Anthropic()

# === Prompt Caching ===
# 将不变的大段内容缓存起来,后续请求只计费增量部分
# 非常适合:长system prompt、RAG上下文、文档分析

# 假设有一个很长的系统提示(如公司规范文档)
long_system_prompt = """
你是XX公司的技术支持AI助手。以下是完整的产品文档:
...(假设这里有10000字的文档)...
"""

# 使用缓存标记
response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": long_system_prompt,
            "cache_control": {"type": "ephemeral"}  # 标记为可缓存
        }
    ],
    messages=[{"role": "user", "content": "产品支持什么操作系统?"}],
)

# 第一次调用:正常计费,创建缓存
# 后续调用:缓存命中,system prompt部分降低90%费用!
print(f"缓存创建token: {response.usage.cache_creation_input_tokens}")
print(f"缓存命中token: {response.usage.cache_read_input_tokens}")

# === RAG 中的缓存使用 ===
# 将检索到的文档放在缓存区,用户后续追问时无需重新计算
rag_context = "从向量数据库检索到的相关文档内容..."

response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[
        {
            "role": "user",
            "content": [
                {
                    "type": "text",
                    "text": f"参考文档:\n{rag_context}",
                    "cache_control": {"type": "ephemeral"}
                },
                {
                    "type": "text",
                    "text": "根据文档回答:产品的定价策略是什么?"
                }
            ]
        }
    ],
)
2

Claude Code — AI 驱动的开发 CLI

安装、命令、Slash Commands、CLAUDE.md、Hooks

2.1 安装与配置

Claude Code 是 Anthropic 官方的命令行编程工具,能直接在终端中与 Claude 协作编写代码、调试、执行命令、管理项目。

Bash
# === 安装 Claude Code ===
npm install -g @anthropic-ai/claude-code

# 验证安装
claude --version

# 启动交互式会话(在项目根目录运行)
cd /path/to/your/project
claude

# === 非交互式使用(脚本/CI集成) ===
# 单次提问
claude -p "这个项目的入口文件是什么?"

# 管道输入
echo "解释这段代码的作用" | claude -p

# 从文件读取提示
cat prompt.txt | claude -p

# 指定输出格式
claude -p "列出所有TODO注释" --output-format json

配置文件层级

位置作用域用途
~/.claude/settings.json全局(所有项目)API密钥、默认模型、全局偏好
.claude/settings.json项目级项目特定配置
CLAUDE.md项目级项目上下文、规范、偏好

2.2 核心能力与工作流

Claude Code 不只是一个聊天框——它能直接读写文件、执行命令、搜索代码库、管理 Git,就像一个结对编程伙伴。

日常开发场景示例

Claude Code 会话
# === 场景1:理解新项目 ===
> 分析这个项目的架构,画出模块依赖关系

# Claude 会自动:
# 1. 用 Glob 扫描项目结构
# 2. 用 Grep 搜索 import 语句
# 3. 用 Read 阅读关键配置文件
# 4. 输出完整的架构分析

# === 场景2:实现功能 ===
> 给用户管理模块添加邮件验证功能

# Claude 会自动:
# 1. 阅读现有用户模块代码
# 2. 理解项目的代码风格和模式
# 3. 编写代码(使用 Edit/Write 工具)
# 4. 根据需要安装依赖
# 5. 运行测试验证

# === 场景3:调试 ===
> 用户登录时返回500错误,帮我定位原因

# Claude 会自动:
# 1. 搜索登录相关代码
# 2. 分析日志(如果可用)
# 3. 发现问题根因
# 4. 提供修复方案

# === 场景4:代码审查 ===
> 审查最近一次 commit 的代码质量和安全性

# Claude 会自动:
# 1. git diff 查看变更
# 2. 分析每个改动
# 3. 指出潜在问题(性能、安全、可读性)

2.3 Slash Commands 斜杠命令

Slash Commands 是 Claude Code 的快捷操作。在会话中输入 / 即可看到可用命令。

命令功能使用示例
/help显示帮助信息查看所有命令
/clear清除对话历史开始新话题时使用
/compact压缩上下文对话太长时压缩
/cost查看当前费用监控API使用成本
/status显示状态查看当前配置
/fast切换快速模式同模型更快输出

自定义 Skill(用户定义的斜杠命令)

你可以创建自己的 Slash Commands,封装常用操作。它们本质上是预定义的 Prompt 模板,存放在 .claude/commands/ 目录下。

Markdown (.claude/commands/review.md)
# 自定义命令: /project:review
# 文件: .claude/commands/review.md

请对当前分支相比于 main 的所有改动进行代码审查:

1. 运行 `git diff main...HEAD` 查看所有变更
2. 逐文件分析,关注以下方面:
   - 代码逻辑正确性
   - 潜在的安全漏洞(注入、XSS、敏感信息泄露)
   - 性能问题(N+1查询、不必要的循环)
   - 错误处理是否完善
   - 代码风格是否一致
3. 给出总体评价和具体建议

输出格式:
## 文件: {filename}
- **问题**: {描述}
- **建议**: {具体修改方案}
- **严重程度**: 🔴高 / 🟡中 / 🟢低
Markdown (.claude/commands/test-gen.md)
# 自定义命令: /project:test-gen
# 文件: .claude/commands/test-gen.md

为 $ARGUMENTS 文件中的所有公开函数/方法生成单元测试:

1. 阅读目标文件,理解每个函数的功能
2. 分析项目现有的测试风格和框架(查看已有测试文件)
3. 生成测试用例,覆盖:
   - 正常输入 (happy path)
   - 边界条件 (boundary cases)
   - 异常输入 (error cases)
   - 空值/null处理
4. 确保测试可以实际运行

使用项目中已有的测试框架和断言风格。
Markdown (~/.claude/commands/daily.md)
# 全局命令: /user:daily
# 文件: ~/.claude/commands/daily.md
# 在任何项目中都可用

今日工作日报生成:
1. 运行 `git log --oneline --since="8 hours ago" --author=$(git config user.name)`
2. 分析今天的代码提交
3. 用以下格式输出:

## 今日完成
- {具体完成的功能/修复}

## 明日计划
- {根据上下文推断的下一步}

## 遇到的问题
- {如果有未解决的问题}
Skill 设计心得
  • Skill 文件支持 $ARGUMENTS 变量,接收用户传入的参数
  • 项目级命令放 .claude/commands/,全局命令放 ~/.claude/commands/
  • 命令名就是文件名(去掉 .md),如 review.md/project:review
  • Skill 本质是 Prompt 工程的封装——把你最佳的 Prompt 固化为可复用的命令

2.4 CLAUDE.md — 项目级AI配置 重要

CLAUDE.md 是 Claude Code 的"项目说明书",放在项目根目录。每次启动 Claude Code 时会自动加载,让 Claude 了解项目上下文。

Markdown (CLAUDE.md)
# CLAUDE.md — 项目上下文配置

## 项目概述
这是一个基于 FastAPI + React 的电商后台管理系统。
- 后端: Python 3.11, FastAPI, SQLAlchemy, PostgreSQL
- 前端: React 18, TypeScript, Ant Design, Vite
- 部署: Docker + Kubernetes

## 代码规范
- Python 使用 black 格式化,isort 排序导入
- TypeScript 使用 ESLint + Prettier
- 所有 API 端点必须有 Pydantic schema 定义
- 数据库操作必须使用 async(异步ORM)
- 不要使用 print(),使用 logging 模块

## 项目结构
```
backend/
├── app/
│   ├── api/          # 路由层:处理 HTTP 请求
│   ├── services/     # 业务逻辑层
│   ├── models/       # SQLAlchemy 数据模型
│   ├── schemas/      # Pydantic 请求/响应 schema
│   └── core/         # 配置、安全、依赖注入
frontend/
├── src/
│   ├── pages/        # 页面组件
│   ├── components/   # 通用组件
│   ├── hooks/        # 自定义 hooks
│   ├── services/     # API 调用层
│   └── stores/       # Zustand 状态管理
```

## 常用命令
- 启动后端: `cd backend && uvicorn app.main:app --reload`
- 启动前端: `cd frontend && npm run dev`
- 运行测试: `cd backend && pytest -v`
- 数据库迁移: `alembic upgrade head`

## 注意事项
- 数据库密码在环境变量 DATABASE_URL 中,不要硬编码
- 所有金额使用 Decimal 类型,不要用 float
- API 返回统一格式: `{"code": 200, "data": ..., "message": "..."}`
- 不要修改 alembic/versions/ 下的已有迁移文件
CLAUDE.md 的威力

有了这个文件,Claude Code 不再是"通用AI",而变成了一个了解你项目的"团队成员"。它会遵循你的代码规范、理解你的架构、使用正确的命令。这就像给新入职的工程师一份详细的 onboarding 文档。

2.5 Hooks — 事件驱动的自动化

Hooks 让你在 Claude Code 的关键操作前后自动执行自定义脚本。

JSON (.claude/settings.json)
{
  "hooks": {
    "PreToolUse": [
      {
        "matcher": "Edit|Write",
        "command": "echo '即将修改文件,请注意审查'"
      }
    ],
    "PostToolUse": [
      {
        "matcher": "Edit|Write",
        "command": "cd backend && python -m py_compile $CLAUDE_FILE_PATH 2>&1 || true"
      }
    ],
    "Notification": [
      {
        "matcher": "",
        "command": "osascript -e 'display notification \"$CLAUDE_NOTIFICATION\" with title \"Claude Code\"'"
      }
    ]
  }
}
Hook 类型触发时机典型用途
PreToolUse工具执行前审批敏感操作、验证参数
PostToolUse工具执行后自动格式化、语法检查、运行lint
NotificationClaude 发送通知时系统通知、声音提醒
Stop会话结束时生成摘要、保存日志
3

Prompt Engineering & Skill 系统设计

高级提示工程、System Prompt设计、Skill封装、思维链

3.1 Prompt Engineering 进阶技巧

结构化 Prompt 模板

高质量的 Prompt 不是随意写的自然语言,而是精心设计的指令系统。以下是经过验证的模板:

Prompt 模板
# === 万能 Prompt 框架 ===

## 角色定义(Role)
你是一位 [具体角色],拥有 [具体专长],在 [领域] 有 [N年] 经验。

## 任务描述(Task)
请完成以下任务:[清晰具体的任务描述]

## 上下文(Context)
背景信息:
- [相关的项目/业务上下文]
- [技术栈/约束条件]
- [之前已做的工作]

## 输出要求(Format)
请按以下格式输出:
- 格式:[Markdown/JSON/代码/表格]
- 语言:[中文/英文]
- 长度:[简洁/详细]
- 必须包含:[必要的元素]
- 不要包含:[需要排除的内容]

## 示例(Examples)- 可选但推荐
输入:[示例输入]
输出:[期望输出]

## 约束(Constraints)
- [必须遵守的规则1]
- [必须遵守的规则2]

实战 Prompt 案例集

Python
# === 案例1:代码审查 Prompt ===
CODE_REVIEW_PROMPT = """你是一位高级代码审查工程师。

请审查以下代码变更,从5个维度评估:

1. **正确性** (0-10): 逻辑是否正确,是否有bug
2. **安全性** (0-10): 是否有注入、XSS、信息泄露等安全隐患
3. **性能** (0-10): 是否有性能问题(N+1查询、内存泄漏、不必要的计算)
4. **可读性** (0-10): 命名是否清晰、结构是否合理、注释是否必要
5. **可维护性** (0-10): 是否便于修改和扩展

对于每个发现的问题,请提供:
- 问题所在行号
- 问题描述
- 修复建议(给出具体代码)
- 严重等级:🔴 必须修复 / 🟡 建议修复 / 🟢 可选优化

代码变更:
```
{diff_content}
```
"""

# === 案例2:SQL 生成 Prompt ===
SQL_GEN_PROMPT = """你是一位SQL专家。根据自然语言描述生成SQL查询。

数据库表结构:
{schema}

规则:
- 使用 PostgreSQL 语法
- 大表查询必须考虑索引利用(给出 EXPLAIN 建议)
- 使用参数化查询($1, $2),不要直接拼接值
- 如果涉及聚合,考虑是否需要窗口函数
- 复杂查询加 SQL 注释解释逻辑

用户需求:{user_request}

请输出:
1. SQL 查询语句
2. 预期执行计划分析
3. 如果查询可能很慢,给出优化建议
"""

# === 案例3:API文档生成 Prompt ===
API_DOC_PROMPT = """分析以下API代码,生成完整的API文档。

代码:
```python
{code}
```

请按以下格式生成文档:

### {HTTP方法} {路径}

**描述**: {功能描述}

**请求参数**:
| 参数名 | 类型 | 必填 | 说明 |
|--------|------|------|------|
| ...    | ...  | ...  | ...  |

**请求示例**:
```json
{示例}
```

**响应示例**:
```json
{示例}
```

**错误码**:
| 状态码 | 说明 |
|--------|------|
| ...    | ...  |
"""

3.2 System Prompt 设计原则 核心

System Prompt 是定义 AI 行为的"宪法"。好的 System Prompt 能让模型输出从80分提升到95分。

Python
# === 生产级 System Prompt 设计 ===

# 1. 客服机器人
CUSTOMER_SERVICE_SYSTEM = """你是「智慧商城」的AI客服助手「小智」。

## 你的身份
- 名称: 小智
- 性格: 友好、专业、耐心
- 语言: 中文,偶尔使用表情增加亲和力

## 能力范围
你可以帮助用户:
✅ 查询订单状态(调用 query_order 工具)
✅ 处理退换货申请(调用 create_return 工具)
✅ 回答产品问题(基于知识库)
✅ 修改收货地址(调用 update_address 工具)

你不能做:
❌ 直接退款(需转人工)
❌ 修改已发货订单
❌ 透露其他用户信息
❌ 回答与商城无关的问题

## 行为规范
1. 始终先确认用户身份(订单号或手机号)
2. 如果不确定答案,说"让我帮您核实一下",而不是猜测
3. 如果问题超出能力范围,礼貌转人工:"这个问题需要人工客服为您处理,我马上转接"
4. 每次对话结束时询问"还有其他可以帮您的吗?"
5. 敏感操作(退货、修改地址)需要用户二次确认

## 回复风格
- 简洁:每条回复不超过3句话
- 结构化:使用编号列表展示多条信息
- 主动:预判用户可能的下一步需求
"""

# 2. 编程助手
CODING_ASSISTANT_SYSTEM = """你是一位资深全栈工程师,精通以下技术栈:
- 后端: Python (FastAPI, Django), Go, Node.js
- 前端: React, Vue, TypeScript
- 数据库: PostgreSQL, Redis, MongoDB
- 基础设施: Docker, K8s, AWS

## 回答原则
1. **先理解再回答**: 如果问题不明确,先提出澄清问题
2. **给出可运行的代码**: 不要给伪代码,给真正能跑的代码
3. **解释关键决策**: 为什么选这个方案,有什么替代方案
4. **考虑生产环境**: 错误处理、日志、性能、安全都要考虑
5. **遵循项目风格**: 如果用户给了现有代码,匹配它的风格

## 代码标准
- 变量命名用英文,注释可以用中文
- 不要过度工程:能简单解决的不要搞复杂
- 不要在回答中加不必要的 type hints 或 docstring
- 安全第一:永远不要建议 eval()、明文密码、关闭 SSL 验证
"""

# 3. 数据分析师
DATA_ANALYST_SYSTEM = """你是一位数据分析专家,擅长:
- 数据探索与清洗
- 统计分析与假设检验
- 可视化设计
- 业务指标解读

## 工作流程
1. 先理解业务问题和数据含义
2. 探索性分析(EDA):数据分布、缺失值、异常值
3. 提出假设并用数据验证
4. 用图表展示关键发现
5. 给出可操作的业务建议

## 输出要求
- 每个分析步骤给出对应的 Python 代码
- 使用 pandas + matplotlib/seaborn
- 数字要有单位和上下文(不要只说"增长20%",要说"环比增长20%,从100万增至120万")
- 区分相关性和因果关系
"""

3.3 Skill 系统设计模式 架构

什么是AI Skill?

Skill 是将特定能力封装为可复用模块的设计模式。一个 Skill = Prompt模板 + 工具集 + 执行逻辑。它让 AI 从"什么都能做但什么都做得一般"变成"特定任务做得极好"。

Python
# === Skill 系统架构 ===
from dataclasses import dataclass, field
from typing import Callable, Any
import json

@dataclass
class Tool:
    """Skill 可用的工具定义"""
    name: str
    description: str
    parameters: dict
    handler: Callable  # 实际执行函数

@dataclass
class Skill:
    """一个可复用的AI技能单元"""
    name: str
    description: str
    system_prompt: str
    tools: list[Tool] = field(default_factory=list)
    examples: list[dict] = field(default_factory=list)  # few-shot 示例
    temperature: float = 0.3
    max_tokens: int = 2048

class SkillEngine:
    """Skill 执行引擎"""

    def __init__(self, client):
        self.client = client
        self.skills: dict[str, Skill] = {}

    def register(self, skill: Skill):
        """注册一个 Skill"""
        self.skills[skill.name] = skill

    def execute(self, skill_name: str, user_input: str, context: dict = None) -> str:
        """执行指定的 Skill"""
        skill = self.skills[skill_name]

        # 构建消息
        messages = []

        # 添加 few-shot 示例
        for example in skill.examples:
            messages.append({"role": "user", "content": example["input"]})
            messages.append({"role": "assistant", "content": example["output"]})

        # 添加上下文(如果有)
        prompt = user_input
        if context:
            prompt = f"上下文信息:\n{json.dumps(context, ensure_ascii=False, indent=2)}\n\n{user_input}"
        messages.append({"role": "user", "content": prompt})

        # 构建工具定义
        tools_def = [{
            "name": t.name,
            "description": t.description,
            "input_schema": t.parameters
        } for t in skill.tools] if skill.tools else None

        # 调用 LLM(含工具循环)
        response = self.client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=skill.max_tokens,
            temperature=skill.temperature,
            system=skill.system_prompt,
            tools=tools_def,
            messages=messages,
        )
        return response.content[0].text


# === 实际 Skill 定义示例 ===

# Skill 1: 代码翻译
code_translator = Skill(
    name="code_translate",
    description="将代码从一种语言翻译到另一种语言",
    system_prompt="""你是一位精通多种编程语言的翻译专家。
将用户提供的源代码翻译为目标语言,保持:
1. 功能完全等价
2. 使用目标语言的惯用写法(不是逐行翻译)
3. 保留原始注释(翻译为中文)
4. 包含必要的导入语句""",
    examples=[
        {
            "input": "将以下Python转为Go:\ndef add(a, b): return a + b",
            "output": "```go\npackage main\n\n// add 返回两个整数的和\nfunc add(a, b int) int {\n\treturn a + b\n}\n```"
        }
    ],
    temperature=0.2,  # 代码翻译需要确定性
)

# Skill 2: SQL解释器
sql_explainer = Skill(
    name="sql_explain",
    description="解释复杂SQL查询的逻辑",
    system_prompt="""你是SQL专家。将复杂SQL查询分解为易于理解的步骤说明。
对每个子查询、JOIN、WHERE条件都给出通俗解释。
最后用一句话总结这个查询的业务含义。""",
    temperature=0.3,
)

# Skill 3: 安全审计
security_audit = Skill(
    name="security_audit",
    description="审计代码的安全漏洞",
    system_prompt="""你是一位应用安全专家(OWASP Top 10)。
检查代码中的安全漏洞,包括但不限于:
- SQL注入 / NoSQL注入
- XSS(跨站脚本)
- CSRF(跨站请求伪造)
- 不安全的反序列化
- 硬编码的密钥/密码
- 路径遍历
- SSRF(服务端请求伪造)

每个发现需要:严重等级、漏洞位置、攻击场景、修复方案""",
    temperature=0.1,  # 安全审计需要严谨
)

3.4 思维链(Chain-of-Thought)与推理增强

Python
# === Extended Thinking — Claude 的深度思考模式 ===
import anthropic

client = anthropic.Anthropic()

# Claude 支持 Extended Thinking:让模型在回答前先进行深度推理
response = client.messages.create(
    model="claude-opus-4-6",
    max_tokens=16000,
    thinking={
        "type": "enabled",
        "budget_tokens": 10000  # 允许模型用最多10000 token来"思考"
    },
    messages=[{
        "role": "user",
        "content": """
        分析以下分布式系统设计的一致性问题:

        系统有3个节点(A, B, C),使用主从复制。
        A是主节点,B和C是从节点。
        当A收到写请求时,先写本地再异步复制到B和C。
        读请求可以发到任意节点。

        问题:
        1. 这个设计可能出现什么一致性问题?
        2. 如何在不牺牲太多性能的情况下改进?
        3. 如果A宕机,如何选举新主节点?
        """
    }]
)

# 查看思考过程和最终答案
for block in response.content:
    if block.type == "thinking":
        print(f"[思考过程]:\n{block.thinking}\n")
    elif block.type == "text":
        print(f"[最终回答]:\n{block.text}")

# === Prompt 技巧:自一致性(Self-Consistency) ===
# 同一个问题多次生成,取多数答案,提高准确率
def self_consistency(question, n_samples=5):
    answers = []
    for _ in range(n_samples):
        resp = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            temperature=0.7,  # 需要一定随机性才能得到不同答案
            messages=[{
                "role": "user",
                "content": f"请一步一步思考,然后在最后一行给出最终答案。\n\n{question}"
            }]
        )
        answers.append(resp.content[0].text)
    # 提取所有答案,取多数
    return answers
4

AI Agent 开发深度指南

Agent架构、ReAct、Multi-Agent、记忆系统、框架对比

4.1 Agent 架构设计 核心

Agent 的核心组件

AI Agent 系统架构 LLM Core(大脑) 推理 Reasoning 规划 Planning 反思 Reflect Memory 🗂 短期:对话上下文 📦 长期:向量数据库 📋 工作:暂存缓冲区 Tools 🔍 搜索 API 💻 代码执行 🌐 HTTP 请求 📁 文件读写 🗄 数据库查询 Environment 📂 文件系统 🗄 数据库 🌏 Web 浏览 🖥 终端
Python
# === 生产级 Agent 框架实现 ===
import anthropic
import json
from typing import Any
from dataclasses import dataclass, field
from abc import ABC, abstractmethod

# --- 工具基类 ---
class BaseTool(ABC):
    name: str
    description: str

    @abstractmethod
    def get_schema(self) -> dict:
        """返回工具的 JSON Schema"""
        pass

    @abstractmethod
    def execute(self, **kwargs) -> str:
        """执行工具并返回结果"""
        pass


# --- 具体工具实现 ---
class WebSearchTool(BaseTool):
    name = "web_search"
    description = "搜索互联网获取最新信息"

    def get_schema(self):
        return {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "搜索关键词"},
                "num_results": {"type": "integer", "default": 5}
            },
            "required": ["query"]
        }

    def execute(self, query: str, num_results: int = 5) -> str:
        # 实际项目中调用搜索 API
        return json.dumps({"results": [f"搜索结果: {query}"]}, ensure_ascii=False)


class PythonExecutor(BaseTool):
    name = "run_python"
    description = "在安全沙箱中执行Python代码,用于数据处理和计算"

    def get_schema(self):
        return {
            "type": "object",
            "properties": {
                "code": {"type": "string", "description": "Python代码"},
                "timeout": {"type": "integer", "default": 30}
            },
            "required": ["code"]
        }

    def execute(self, code: str, timeout: int = 30) -> str:
        import subprocess
        try:
            result = subprocess.run(
                ["python", "-c", code],
                capture_output=True, text=True, timeout=timeout
            )
            return result.stdout or result.stderr or "执行完成,无输出"
        except subprocess.TimeoutExpired:
            return "执行超时"
        except Exception as e:
            return f"执行错误: {e}"


class FileManager(BaseTool):
    name = "file_ops"
    description = "读取或写入文件"

    def get_schema(self):
        return {
            "type": "object",
            "properties": {
                "action": {"type": "string", "enum": ["read", "write", "list"]},
                "path": {"type": "string"},
                "content": {"type": "string", "description": "写入内容(action=write时必需)"}
            },
            "required": ["action", "path"]
        }

    def execute(self, action: str, path: str, content: str = "") -> str:
        import os
        if action == "read":
            with open(path, 'r') as f:
                return f.read()[:5000]  # 限制长度
        elif action == "write":
            with open(path, 'w') as f:
                f.write(content)
            return f"文件已写入: {path}"
        elif action == "list":
            items = os.listdir(path)
            return "\n".join(items[:50])
        return "未知操作"


# --- Agent 核心 ---
class Agent:
    """生产级 AI Agent"""

    def __init__(self, name: str, system_prompt: str, tools: list[BaseTool],
                 model: str = "claude-sonnet-4-6", max_iterations: int = 10):
        self.name = name
        self.client = anthropic.Anthropic()
        self.model = model
        self.system_prompt = system_prompt
        self.tools = {t.name: t for t in tools}
        self.max_iterations = max_iterations
        self.conversation_history = []

    def _get_tools_config(self):
        return [{
            "name": t.name,
            "description": t.description,
            "input_schema": t.get_schema()
        } for t in self.tools.values()]

    def run(self, task: str) -> str:
        """执行任务的主循环"""
        self.conversation_history = [
            {"role": "user", "content": task}
        ]

        for iteration in range(self.max_iterations):
            response = self.client.messages.create(
                model=self.model,
                max_tokens=4096,
                system=self.system_prompt,
                tools=self._get_tools_config(),
                messages=self.conversation_history,
            )

            # 处理响应
            if response.stop_reason == "end_turn":
                # Agent 完成任务
                final_text = ""
                for block in response.content:
                    if hasattr(block, 'text'):
                        final_text += block.text
                return final_text

            elif response.stop_reason == "tool_use":
                # 需要调用工具
                self.conversation_history.append({
                    "role": "assistant",
                    "content": response.content
                })

                tool_results = []
                for block in response.content:
                    if block.type == "tool_use":
                        tool = self.tools.get(block.name)
                        if tool:
                            print(f"  [{self.name}] 调用工具: {block.name}")
                            try:
                                result = tool.execute(**block.input)
                            except Exception as e:
                                result = f"工具执行失败: {e}"
                        else:
                            result = f"未知工具: {block.name}"

                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": str(result)[:10000],  # 限制结果长度
                        })

                self.conversation_history.append({
                    "role": "user",
                    "content": tool_results
                })

        return "达到最大迭代次数,任务未完成"

# === 创建并运行 Agent ===
agent = Agent(
    name="研究助手",
    system_prompt="""你是一个研究助手 Agent。你可以搜索网络、执行代码、读写文件。

工作流程:
1. 分析用户的研究问题
2. 搜索相关信息
3. 用Python处理和分析数据
4. 将研究结果写入文件
5. 给出结构化的研究报告

每一步都要说明你在做什么以及为什么。""",
    tools=[WebSearchTool(), PythonExecutor(), FileManager()],
)

# result = agent.run("调研2024年全球AI市场规模和增长趋势,生成分析报告")

4.2 ReAct 模式详解

ReAct = Reasoning + Acting,是当前最主流的 Agent 执行模式。核心思想是让 LLM 交替进行"思考"和"行动"。

ReAct 执行流程

Task: "查询苹果公司最新市值并与微软对比"

Thought 1: 我需要搜索苹果公司最新市值
Action 1:  web_search("Apple market cap 2025")
Observation 1: Apple市值约3.5万亿美元...

Thought 2: 现在需要搜索微软市值
Action 2:  web_search("Microsoft market cap 2025")
Observation 2: Microsoft市值约3.2万亿美元...

Thought 3: 我有了两家公司的数据,可以对比分析了
Action 3:  run_python("...")  # 计算差异、生成图表
Observation 3: 图表已生成...

Thought 4: 信息收集完毕,可以给出最终答案
Final Answer: 苹果市值3.5万亿美元,高于微软的3.2万亿...
Claude 的 Tool Use 天然就是 ReAct

Claude API 的 Tool Use 机制已经内建了 ReAct 模式——Claude 会先思考(在text中解释推理过程),然后决定调用工具(tool_use),你返回结果后它继续推理。不需要额外框架来实现。

4.3 Multi-Agent 多智能体协作 前沿

单个 Agent 能力有限,Multi-Agent 让多个专业 Agent 协作完成复杂任务。

Python
# === Multi-Agent 协作框架 ===
import anthropic
from enum import Enum

class AgentRole(Enum):
    PLANNER = "planner"       # 规划者:拆解任务
    RESEARCHER = "researcher" # 研究者:搜集信息
    CODER = "coder"           # 编码者:编写代码
    REVIEWER = "reviewer"     # 审查者:质量把关
    ORCHESTRATOR = "orchestrator"  # 协调者:管理流程

class MultiAgentSystem:
    """多Agent协作系统"""

    def __init__(self):
        self.client = anthropic.Anthropic()
        self.agents = {}
        self.message_bus = []  # Agent 间的消息总线

    def add_agent(self, role: AgentRole, system_prompt: str, tools=None):
        self.agents[role] = {
            "system_prompt": system_prompt,
            "tools": tools or [],
            "history": []
        }

    def _call_agent(self, role: AgentRole, message: str) -> str:
        """调用指定 Agent"""
        agent = self.agents[role]
        agent["history"].append({"role": "user", "content": message})

        response = self.client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=4096,
            system=agent["system_prompt"],
            messages=agent["history"],
        )
        reply = response.content[0].text
        agent["history"].append({"role": "assistant", "content": reply})
        return reply

    def execute_pipeline(self, task: str) -> dict:
        """执行多Agent流水线"""
        results = {}

        # Step 1: 规划者拆解任务
        print("[Planner] 分析和拆解任务...")
        plan = self._call_agent(AgentRole.PLANNER,
            f"请分析以下任务并拆解为具体的子任务:\n{task}\n\n"
            "输出JSON格式:[{\"id\": 1, \"task\": \"...\", \"assignee\": \"researcher|coder\"}]"
        )
        results["plan"] = plan

        # Step 2: 研究者收集信息
        print("[Researcher] 收集相关信息...")
        research = self._call_agent(AgentRole.RESEARCHER,
            f"根据以下任务计划,收集所需信息:\n{plan}"
        )
        results["research"] = research

        # Step 3: 编码者实现
        print("[Coder] 编写代码...")
        code = self._call_agent(AgentRole.CODER,
            f"任务: {task}\n\n参考信息: {research}\n\n请编写完整可运行的代码。"
        )
        results["code"] = code

        # Step 4: 审查者把关
        print("[Reviewer] 代码审查...")
        review = self._call_agent(AgentRole.REVIEWER,
            f"请审查以下代码的质量、安全性和性能:\n{code}"
        )
        results["review"] = review

        return results


# === 创建多Agent系统 ===
system = MultiAgentSystem()

system.add_agent(AgentRole.PLANNER,
    "你是一个项目经理,擅长将复杂任务拆解为清晰的子任务。输出JSON格式。")

system.add_agent(AgentRole.RESEARCHER,
    "你是一个技术研究员,擅长收集技术方案、最佳实践和相关文档。")

system.add_agent(AgentRole.CODER,
    "你是一位高级开发工程师,编写高质量、可维护的代码。始终包含错误处理和注释。")

system.add_agent(AgentRole.REVIEWER,
    "你是一位代码审查专家,关注安全漏洞、性能问题、代码规范。给出具体修改建议。")

# result = system.execute_pipeline("开发一个基于FastAPI的用户认证系统,支持JWT和OAuth2")

4.4 Agent 记忆系统

Python
# === 三层记忆架构 ===
import json
from datetime import datetime
from collections import deque

class AgentMemory:
    """Agent 的记忆系统"""

    def __init__(self, short_term_limit=20, embed_model=None):
        # 1. 短期记忆:最近的对话(滑动窗口)
        self.short_term = deque(maxlen=short_term_limit)

        # 2. 工作记忆:当前任务的暂存信息
        self.working_memory = {}

        # 3. 长期记忆:向量数据库存储的持久知识
        self.embed_model = embed_model
        self.long_term_store = []  # 实际项目中用ChromaDB/Pinecone

    def add_conversation(self, role: str, content: str):
        """添加对话到短期记忆"""
        self.short_term.append({
            "role": role,
            "content": content,
            "timestamp": datetime.now().isoformat()
        })

    def set_working(self, key: str, value: any):
        """设置工作记忆"""
        self.working_memory[key] = value

    def get_working(self, key: str, default=None):
        """获取工作记忆"""
        return self.working_memory.get(key, default)

    def save_to_long_term(self, content: str, metadata: dict = None):
        """保存到长期记忆"""
        entry = {
            "content": content,
            "metadata": metadata or {},
            "timestamp": datetime.now().isoformat(),
        }
        self.long_term_store.append(entry)
        # 实际实现:embedding + 向量数据库存储

    def search_long_term(self, query: str, top_k: int = 3) -> list:
        """搜索长期记忆"""
        # 实际实现:向量相似度搜索
        # 这里简化为关键词匹配
        results = []
        for entry in self.long_term_store:
            if any(word in entry["content"] for word in query.split()):
                results.append(entry)
        return results[:top_k]

    def get_context_for_llm(self) -> str:
        """构建发给LLM的记忆上下文"""
        parts = []

        if self.working_memory:
            parts.append("## 当前任务信息")
            for k, v in self.working_memory.items():
                parts.append(f"- {k}: {v}")

        if self.long_term_store:
            parts.append("\n## 相关历史知识")
            for entry in self.long_term_store[-3:]:
                parts.append(f"- {entry['content'][:200]}")

        return "\n".join(parts)

# === 使用示例 ===
memory = AgentMemory()
memory.set_working("current_task", "构建用户认证系统")
memory.set_working("tech_stack", "FastAPI + PostgreSQL + JWT")
memory.save_to_long_term(
    "上次用户认证项目中,JWT过期时间设为30分钟效果最好",
    {"project": "auth-v1", "type": "lesson_learned"}
)

context = memory.get_context_for_llm()
print(context)

4.5 Agent 框架对比 选型

框架定位优点缺点适用场景
LangChain 通用LLM框架 生态最丰富,集成最多 过度抽象,学习曲线陡 需要快速集成多种服务
LlamaIndex RAG专精框架 数据连接最强,索引优化好 Agent能力弱于LangChain 知识库/文档问答系统
CrewAI 多Agent协作 简单直观,角色定义清晰 灵活度有限 需要多角色协作的任务
AutoGen (微软) Agent对话框架 对话式协作,可编程性强 文档不够完善 Agent间对话式协作
Claude Agent SDK Anthropic原生 与Claude深度集成 仅限Claude模型 基于Claude的Agent开发
自建 (推荐入门) 学习/定制 完全控制,理解原理 需要自行实现基础功能 学习阶段 / 定制需求强
选型建议

如果你是 AI Agent 新手,建议先用原生 API 自建一个简单 Agent(如上面的代码),理解核心原理后再使用框架。直接上框架容易"知其然不知其所以然"。对于生产项目,根据具体需求选择框架或自建。

5

RAG 系统进阶

高级检索、向量数据库、Evaluation

5.1 高级检索策略 核心

RAG 进阶架构 用户查询 查询改写 HyDE · 多查询 · 步退式 向量检索 语义匹配 关键词检索 BM25 精确 知识图谱 结构化关系 重排序(Re-ranking) Cross-Encoder · Cohere Rerank 上下文压缩 过滤不相关段落 LLM 生成(带引用回答)
Python
# === 完整的生产级 RAG 实现 ===
import chromadb
from sentence_transformers import SentenceTransformer
from typing import Optional
import hashlib
import re

class ProductionRAG:
    """生产级 RAG 系统"""

    def __init__(self, collection_name="knowledge_base"):
        self.embed_model = SentenceTransformer("all-MiniLM-L6-v2")
        self.chroma_client = chromadb.PersistentClient(path="./chroma_db")
        self.collection = self.chroma_client.get_or_create_collection(
            name=collection_name,
            metadata={"hnsw:space": "cosine"}  # 使用余弦相似度
        )

    # --- 文档处理 ---
    def chunk_document(self, text: str, chunk_size=500, overlap=100,
                       metadata: dict = None) -> list[dict]:
        """智能分块:按段落优先,保持语义完整性"""
        # 先按段落分割
        paragraphs = re.split(r'\n\s*\n', text)

        chunks = []
        current_chunk = ""

        for para in paragraphs:
            if len(current_chunk) + len(para) <= chunk_size:
                current_chunk += para + "\n\n"
            else:
                if current_chunk:
                    chunks.append({
                        "text": current_chunk.strip(),
                        "id": hashlib.md5(current_chunk.encode()).hexdigest()[:12],
                        "metadata": metadata or {}
                    })
                current_chunk = para + "\n\n"

        if current_chunk.strip():
            chunks.append({
                "text": current_chunk.strip(),
                "id": hashlib.md5(current_chunk.encode()).hexdigest()[:12],
                "metadata": metadata or {}
            })

        return chunks

    def ingest(self, documents: list[dict]):
        """批量导入文档"""
        texts = [d["text"] for d in documents]
        ids = [d["id"] for d in documents]
        metadatas = [d.get("metadata", {}) for d in documents]

        self.collection.add(
            documents=texts,
            ids=ids,
            metadatas=metadatas,
        )
        print(f"已导入 {len(documents)} 个文档块")

    # --- 检索策略 ---
    def retrieve(self, query: str, top_k: int = 5,
                 filters: dict = None) -> list[dict]:
        """基础向量检索"""
        where = filters if filters else None
        results = self.collection.query(
            query_texts=[query],
            n_results=top_k,
            where=where,
        )
        return [{
            "text": doc,
            "score": 1 - dist,  # chroma返回距离,转为相似度
            "metadata": meta,
        } for doc, dist, meta in zip(
            results["documents"][0],
            results["distances"][0],
            results["metadatas"][0]
        )]

    def hybrid_retrieve(self, query: str, top_k: int = 5) -> list[dict]:
        """混合检索:向量 + 关键词"""
        # 向量检索
        vector_results = self.retrieve(query, top_k=top_k * 2)

        # 关键词加权(简化版,实际可用BM25)
        keywords = set(query.lower().split())
        for result in vector_results:
            keyword_score = sum(
                1 for kw in keywords if kw in result["text"].lower()
            ) / max(len(keywords), 1)
            # 融合分数:70% 向量 + 30% 关键词
            result["hybrid_score"] = 0.7 * result["score"] + 0.3 * keyword_score

        # 按混合分数排序
        vector_results.sort(key=lambda x: x["hybrid_score"], reverse=True)
        return vector_results[:top_k]

    def hyde_retrieve(self, query: str, llm_client, top_k: int = 5) -> list[dict]:
        """HyDE (Hypothetical Document Embeddings)
        先让LLM生成一个假设性的回答,用回答去检索,而非用问题"""
        response = llm_client.messages.create(
            model="claude-haiku-4-5-20251001",
            max_tokens=300,
            messages=[{
                "role": "user",
                "content": f"请直接回答以下问题(不需要说明你不确定):\n{query}"
            }]
        )
        hypothetical_doc = response.content[0].text
        # 用假设性回答进行检索(通常效果更好)
        return self.retrieve(hypothetical_doc, top_k=top_k)

    # --- 生成 ---
    def answer(self, query: str, llm_client, top_k: int = 5) -> dict:
        """完整的 RAG 问答流程"""
        # 1. 检索
        retrieved = self.hybrid_retrieve(query, top_k=top_k)

        # 2. 构建上下文
        context_parts = []
        for i, doc in enumerate(retrieved, 1):
            source = doc["metadata"].get("source", "未知")
            context_parts.append(f"[来源{i}: {source}]\n{doc['text']}")
        context = "\n\n---\n\n".join(context_parts)

        # 3. 生成回答
        response = llm_client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=2048,
            system="""基于提供的参考资料回答用户问题。
规则:
1. 只使用参考资料中的信息,不要编造
2. 如果资料中没有相关信息,明确说明
3. 在回答中用 [来源N] 标注信息出处
4. 保持回答简洁准确""",
            messages=[{
                "role": "user",
                "content": f"参考资料:\n{context}\n\n用户问题:{query}"
            }]
        )

        return {
            "answer": response.content[0].text,
            "sources": retrieved,
            "tokens_used": response.usage.input_tokens + response.usage.output_tokens
        }

# === 使用示例 ===
rag = ProductionRAG()

# 导入文档
docs = rag.chunk_document(
    """FastAPI 是一个现代的Python Web框架...(长文档)...""",
    metadata={"source": "fastapi_docs.md", "category": "backend"}
)
rag.ingest(docs)

# 问答
# result = rag.answer("FastAPI如何处理异步请求?", llm_client)
# print(result["answer"])

5.2 向量数据库选型

数据库类型特点适用场景
ChromaDB嵌入式零配置,Python原生,适合原型开发/小规模
Pinecone云托管全托管,无需运维,自动扩展快速上线的SaaS产品
Milvus分布式高性能,支持十亿级向量大规模生产环境
Weaviate全功能内置向量化,支持混合搜索需要GraphQL查询
Qdrant高性能Rust编写,过滤性能极强需要复杂过滤条件
pgvectorPG扩展在现有PG上加向量能力已有PG基础设施
FAISS库(非数据库)Meta出品,纯计算最快离线批量检索

5.3 RAG 系统评估

Python
# === RAG 评估框架 ===
# 评估 RAG 系统需要关注两个维度:检索质量 + 生成质量

class RAGEvaluator:
    """RAG 系统评估"""

    def __init__(self, llm_client):
        self.client = llm_client

    def evaluate_retrieval(self, query: str, retrieved_docs: list,
                           ground_truth_docs: list) -> dict:
        """评估检索质量"""
        retrieved_set = set(d["id"] for d in retrieved_docs)
        truth_set = set(d["id"] for d in ground_truth_docs)

        # 精确率:检索到的中有多少是相关的
        precision = len(retrieved_set & truth_set) / max(len(retrieved_set), 1)
        # 召回率:相关的中有多少被检索到
        recall = len(retrieved_set & truth_set) / max(len(truth_set), 1)
        # F1
        f1 = 2 * precision * recall / max(precision + recall, 1e-10)

        return {"precision": precision, "recall": recall, "f1": f1}

    def evaluate_answer(self, question: str, answer: str,
                        reference: str) -> dict:
        """用LLM评估回答质量(LLM-as-Judge)"""
        eval_prompt = f"""请评估AI回答的质量。满分5分。

问题:{question}
参考答案:{reference}
AI回答:{answer}

请从以下维度评分(1-5分):
1. 准确性:回答是否正确
2. 完整性:是否覆盖了关键信息
3. 相关性:是否紧扣问题
4. 简洁性:是否简洁不冗余

输出JSON:{{"accuracy": N, "completeness": N, "relevance": N, "conciseness": N, "overall": N, "reason": "..."}}"""

        response = self.client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=500,
            messages=[{"role": "user", "content": eval_prompt}]
        )
        return response.content[0].text  # 解析JSON
6

开源生态 & 框架实战

LangChain、LlamaIndex、CrewAI、开源LLM部署

6.1 LangChain — 最流行的LLM框架

Python
# pip install langchain langchain-anthropic langchain-community

from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser

# === 基础使用 ===
llm = ChatAnthropic(model="claude-sonnet-4-6", temperature=0.3)

# 简单调用
response = llm.invoke([HumanMessage(content="Hello!")])
print(response.content)

# === 链式调用(Chain) — LangChain 的核心概念 ===
# Prompt → LLM → 输出解析器,用管道符 | 连接
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个{role}专家。"),
    ("user", "{question}")
])

chain = prompt | llm  # 管道语法:prompt → llm
response = chain.invoke({"role": "Python", "question": "解释装饰器"})

# === 带工具的 Agent ===
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain.tools import tool

@tool
def calculator(expression: str) -> str:
    """计算数学表达式的值"""
    try:
        return str(eval(expression))
    except:
        return "计算错误"

@tool
def get_current_time() -> str:
    """获取当前时间"""
    from datetime import datetime
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

# 创建 Agent
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有帮助的助手,可以使用工具回答问题。"),
    ("placeholder", "{chat_history}"),
    ("user", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

agent = create_tool_calling_agent(llm, [calculator, get_current_time], prompt)
executor = AgentExecutor(agent=agent, tools=[calculator, get_current_time], verbose=True)

# result = executor.invoke({"input": "现在几点了?另外帮我算 (15 * 23) + 47", "chat_history": []})

# === LangChain RAG 链 ===
from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_core.runnables import RunnablePassthrough

# 文档分割
splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=100)
# docs = splitter.split_documents(loaded_docs)

# 向量存储
embeddings = HuggingFaceEmbeddings(model_name="all-MiniLM-L6-v2")
# vectorstore = Chroma.from_documents(docs, embeddings)
# retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

# RAG 链
rag_prompt = ChatPromptTemplate.from_template("""
基于以下上下文回答问题:
{context}

问题:{question}
""")

# rag_chain = (
#     {"context": retriever, "question": RunnablePassthrough()}
#     | rag_prompt
#     | llm
# )
# answer = rag_chain.invoke("什么是Transformer?")

6.2 LlamaIndex — RAG 专精框架

Python
# pip install llama-index llama-index-llms-anthropic

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, Settings
from llama_index.llms.anthropic import Anthropic

# 配置 LLM
Settings.llm = Anthropic(model="claude-sonnet-4-6")

# === 5行代码构建一个文档问答系统 ===
# 1. 加载文档(支持 PDF, DOCX, TXT, HTML 等)
# documents = SimpleDirectoryReader("./data").load_data()

# 2. 构建索引(自动分块 + 向量化 + 存储)
# index = VectorStoreIndex.from_documents(documents)

# 3. 创建查询引擎
# query_engine = index.as_query_engine(similarity_top_k=5)

# 4. 查询
# response = query_engine.query("这个项目的架构是什么?")
# print(response)

# === 高级功能:子问题查询引擎 ===
# 自动将复杂问题拆分为子问题,分别查询后综合回答
from llama_index.core.query_engine import SubQuestionQueryEngine
from llama_index.core.tools import QueryEngineTool, ToolMetadata

# 为不同的数据源创建查询引擎
# api_engine = api_index.as_query_engine()
# arch_engine = arch_index.as_query_engine()

# query_engine_tools = [
#     QueryEngineTool(
#         query_engine=api_engine,
#         metadata=ToolMetadata(name="api_docs", description="API接口文档"),
#     ),
#     QueryEngineTool(
#         query_engine=arch_engine,
#         metadata=ToolMetadata(name="arch_docs", description="架构设计文档"),
#     ),
# ]

# 子问题引擎会自动决定查哪个数据源
# engine = SubQuestionQueryEngine.from_defaults(query_engine_tools=query_engine_tools)
# response = engine.query("这个系统的认证API是如何设计的?遵循了什么架构原则?")

6.3 CrewAI — 多Agent协作框架

Python
# pip install crewai crewai-tools

from crewai import Agent, Task, Crew, Process

# === CrewAI 的核心概念 ===
# Agent = 一个有角色、目标、工具的AI实体
# Task = 一个具体的任务,分配给某个Agent
# Crew = 一组Agent,协作完成所有Task

# 定义 Agent
researcher = Agent(
    role="技术研究员",
    goal="深入研究指定技术主题,收集全面准确的信息",
    backstory="你是一位资深技术分析师,擅长从各种来源收集和综合技术信息。",
    verbose=True,
    llm="claude-sonnet-4-6",
)

writer = Agent(
    role="技术写作专家",
    goal="将研究结果编写成清晰、专业的技术文档",
    backstory="你是一位技术博客作者,擅长将复杂概念用简洁的语言解释。",
    verbose=True,
    llm="claude-sonnet-4-6",
)

reviewer = Agent(
    role="内容审核专家",
    goal="审查技术文档的准确性和质量",
    backstory="你是一位严谨的技术编辑,对内容质量有极高要求。",
    verbose=True,
    llm="claude-sonnet-4-6",
)

# 定义任务
research_task = Task(
    description="研究 {topic} 的最新发展、核心概念和实际应用场景",
    expected_output="包含关键发现、数据支撑和参考来源的研究报告",
    agent=researcher,
)

writing_task = Task(
    description="基于研究报告编写一篇2000字的技术博文",
    expected_output="结构清晰、内容准确的技术博客文章",
    agent=writer,
    context=[research_task],  # 依赖研究任务的结果
)

review_task = Task(
    description="审查博文的技术准确性、可读性和完整性",
    expected_output="审核意见和最终修改后的文章",
    agent=reviewer,
    context=[writing_task],
)

# 组建团队
crew = Crew(
    agents=[researcher, writer, reviewer],
    tasks=[research_task, writing_task, review_task],
    process=Process.sequential,  # 顺序执行(也支持 hierarchical)
    verbose=True,
)

# 执行
# result = crew.kickoff(inputs={"topic": "AI Agent 技术栈 2025"})
# print(result)

6.4 开源LLM本地部署

模型参数量特点推荐部署方式
LLaMA 3.18B/70B/405BMeta开源标杆vLLM / Ollama
Qwen 2.57B/72B阿里通义,中文强vLLM
DeepSeek V3671B (MoE)MoE架构,性价比高多GPU + vLLM
Mistral7B高效小模型Ollama
Phi-33.8B微软小模型,推理强Ollama
Bash
# === Ollama — 最简单的本地LLM运行方式 ===
# 安装(macOS/Linux)
curl -fsSL https://ollama.ai/install.sh | sh

# 运行模型(自动下载)
ollama run llama3.1:8b        # LLaMA 3.1 8B
ollama run qwen2.5:7b         # 通义千问 2.5
ollama run deepseek-r1:8b     # DeepSeek R1 蒸馏版

# Ollama 兼容 OpenAI API
# 基础 URL: http://localhost:11434/v1
Python
# === 用 OpenAI SDK 调用本地模型 ===
from openai import OpenAI

# Ollama 本地模型
client = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")
response = client.chat.completions.create(
    model="llama3.1:8b",
    messages=[{"role": "user", "content": "Hello!"}],
)
print(response.choices[0].message.content)

# vLLM 部署的模型
# python -m vllm.entrypoints.openai.api_server --model meta-llama/Llama-3.1-8B-Instruct
client = OpenAI(base_url="http://localhost:8000/v1", api_key="dummy")
# 使用方式完全相同
7

MCP 协议 — Model Context Protocol

Anthropic 提出的 AI 工具交互标准协议

7.1 MCP 概述 新标准

什么是 MCP?

Model Context Protocol (MCP) 是 Anthropic 提出的开放协议,为 AI 模型与外部工具/数据源之间的交互定义了标准接口。可以把它理解为 AI 世界的"USB协议"——不管是什么工具,只要遵循 MCP 协议,就能被任何支持 MCP 的 AI 应用使用。

MCP 架构 MCP Client 💻 Claude Code 🖥 Claude Desktop 🏗 自建 AI 应用 MCP 协议 JSON-RPC stdio / SSE MCP Server 你开发的工具服务 🔧 Tools(工具) 📦 Resources(资源) 📝 Prompts(提示) 遵循 MCP 标准接口 已有 MCP Server 生态 📁 文件系统 🐙 GitHub 💬 Slack 🐘 PostgreSQL ☁️ Google Drive 📋 Jira 🌐 Puppeteer ✨ 自定义 Server 遵循 MCP 协议的任何工具都可接入任何 MCP 客户端

MCP 的三大能力

能力说明类比
Tools模型可以调用的函数/操作API 端点
Resources模型可以读取的数据源文件/数据库
Prompts预定义的提示模板Slash Commands

7.2 开发 MCP Server 实战

Python
# === 从零开发一个 MCP Server ===
# pip install mcp

from mcp.server.fastmcp import FastMCP
import json
import sqlite3

# 创建 MCP Server
mcp = FastMCP("my-tools-server")

# === 定义 Tool ===
@mcp.tool()
def query_database(sql: str) -> str:
    """执行 SQL 查询并返回结果。仅支持 SELECT 语句。

    Args:
        sql: SQL查询语句(仅限SELECT)
    """
    if not sql.strip().upper().startswith("SELECT"):
        return "错误:仅允许 SELECT 查询"

    conn = sqlite3.connect("app.db")
    try:
        cursor = conn.execute(sql)
        columns = [desc[0] for desc in cursor.description]
        rows = cursor.fetchall()
        results = [dict(zip(columns, row)) for row in rows]
        return json.dumps(results, ensure_ascii=False, indent=2)
    except Exception as e:
        return f"查询错误: {e}"
    finally:
        conn.close()

@mcp.tool()
def send_notification(channel: str, message: str) -> str:
    """发送通知消息到指定渠道。

    Args:
        channel: 通知渠道 (email/slack/webhook)
        message: 通知内容
    """
    # 实际项目中调用通知服务
    print(f"[{channel}] {message}")
    return f"通知已发送到 {channel}"

@mcp.tool()
def analyze_logs(service: str, time_range: str = "1h") -> str:
    """分析指定服务的日志,返回错误摘要。

    Args:
        service: 服务名称
        time_range: 时间范围 (如 '1h', '24h', '7d')
    """
    # 模拟日志分析
    return json.dumps({
        "service": service,
        "time_range": time_range,
        "total_requests": 15234,
        "errors": 23,
        "error_rate": "0.15%",
        "top_errors": [
            {"code": 500, "count": 12, "message": "Database connection timeout"},
            {"code": 429, "count": 8, "message": "Rate limit exceeded"},
        ]
    }, ensure_ascii=False, indent=2)

# === 定义 Resource ===
@mcp.resource("config://app")
def get_app_config() -> str:
    """获取应用配置信息"""
    return json.dumps({
        "app_name": "MyApp",
        "version": "2.1.0",
        "environment": "production",
        "features": {"dark_mode": True, "beta_features": False}
    })

@mcp.resource("metrics://dashboard")
def get_metrics() -> str:
    """获取系统监控指标"""
    return json.dumps({
        "cpu_usage": "45%",
        "memory_usage": "62%",
        "active_connections": 1234,
        "response_time_p99": "230ms"
    })

# === 定义 Prompt ===
@mcp.prompt()
def debug_prompt(error_message: str) -> str:
    """生成调试提示"""
    return f"""请帮我调试以下错误:

错误信息:{error_message}

请按以下步骤分析:
1. 错误类型和可能原因
2. 查询相关日志(使用 analyze_logs 工具)
3. 检查相关配置(使用 config://app 资源)
4. 给出具体的修复建议"""

# 启动服务器
if __name__ == "__main__":
    mcp.run()

# === 启动方式 ===
# 1. stdio 模式(Claude Code 使用):
#    python my_mcp_server.py
#
# 2. SSE 模式(Web应用使用):
#    python my_mcp_server.py --transport sse --port 3001

在 Claude Code 中配置 MCP Server

JSON (~/.claude/settings.json)
{
  "mcpServers": {
    "my-tools": {
      "command": "python",
      "args": ["/path/to/my_mcp_server.py"],
      "env": {
        "DATABASE_URL": "sqlite:///app.db"
      }
    },
    "github": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-github"],
      "env": {
        "GITHUB_TOKEN": "ghp_xxxx"
      }
    },
    "filesystem": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-filesystem", "/path/to/allowed/dir"]
    }
  }
}

7.3 MCP 实战案例

案例:运维监控 Agent

Python
# === 基于 MCP 的运维 Agent ===
# 这个 MCP Server 让 Claude 能够监控和管理服务器

from mcp.server.fastmcp import FastMCP
import subprocess
import json
import psutil

mcp = FastMCP("ops-agent")

@mcp.tool()
def check_system_health() -> str:
    """检查系统健康状态:CPU、内存、磁盘"""
    return json.dumps({
        "cpu_percent": psutil.cpu_percent(interval=1),
        "memory": {
            "total_gb": round(psutil.virtual_memory().total / 1e9, 1),
            "used_percent": psutil.virtual_memory().percent,
        },
        "disk": {
            "total_gb": round(psutil.disk_usage('/').total / 1e9, 1),
            "used_percent": round(psutil.disk_usage('/').percent, 1),
        },
        "load_avg": list(psutil.getloadavg()),
    }, indent=2)

@mcp.tool()
def list_processes(sort_by: str = "memory") -> str:
    """列出资源占用最高的进程

    Args:
        sort_by: 排序方式 (memory/cpu)
    """
    procs = []
    for p in psutil.process_iter(['pid', 'name', 'cpu_percent', 'memory_percent']):
        info = p.info
        if info['memory_percent'] and info['memory_percent'] > 0.1:
            procs.append(info)

    key = 'memory_percent' if sort_by == 'memory' else 'cpu_percent'
    procs.sort(key=lambda x: x.get(key, 0), reverse=True)

    return json.dumps(procs[:15], indent=2)

@mcp.tool()
def check_docker_containers() -> str:
    """列出所有 Docker 容器的状态"""
    try:
        result = subprocess.run(
            ["docker", "ps", "-a", "--format", "{{json .}}"],
            capture_output=True, text=True
        )
        containers = [json.loads(line) for line in result.stdout.strip().split('\n') if line]
        return json.dumps(containers, indent=2)
    except Exception as e:
        return f"Docker 不可用: {e}"

@mcp.tool()
def tail_logs(service: str, lines: int = 50) -> str:
    """查看服务日志的最后N行

    Args:
        service: 服务/容器名称
        lines: 查看行数
    """
    try:
        result = subprocess.run(
            ["docker", "logs", "--tail", str(lines), service],
            capture_output=True, text=True
        )
        return result.stdout or result.stderr or "无日志输出"
    except Exception as e:
        return f"获取日志失败: {e}"

if __name__ == "__main__":
    mcp.run()

# 配置到Claude Code后,你可以这样使用:
# > 检查一下服务器状态,有没有异常的进程?
# > 查看 web-api 容器的最近日志,有没有报错?
# > 系统内存使用率是多少?哪些进程占用最多?
8

实战项目案例

企业知识库、代码审查Agent、数据分析助手、多Agent系统

8.1 项目一:企业知识库问答系统 完整项目

企业知识库系统架构 用户界面 React / Streamlit HTTP API FastAPI 后端 文档管理 📄 上传 & 解析 ✂️ 分块策略 🔢 向量化嵌入 💾 入库存储 问答引擎 ✍️ 查询改写 🔀 混合检索 📊 Rerank 重排 🤖 LLM 生成 + 引用 🗄 ChromaDB / Milvus(向量数据库)
Python
# === 企业知识库完整后端 ===
from fastapi import FastAPI, UploadFile, File, HTTPException
from pydantic import BaseModel
import anthropic
import chromadb
from sentence_transformers import SentenceTransformer
import hashlib
import re

app = FastAPI(title="Enterprise Knowledge Base")
client = anthropic.Anthropic()
embed_model = SentenceTransformer("all-MiniLM-L6-v2")
chroma = chromadb.PersistentClient(path="./kb_data")
collection = chroma.get_or_create_collection("knowledge_base")

# --- 数据模型 ---
class QuestionRequest(BaseModel):
    question: str
    top_k: int = 5
    filters: dict = None

class AnswerResponse(BaseModel):
    answer: str
    sources: list
    confidence: float

# --- 文档处理 ---
def smart_chunk(text: str, source: str, chunk_size=800, overlap=200):
    """智能分块:优先按标题分割"""
    sections = re.split(r'\n#{1,3}\s+', text)
    chunks = []
    for i, section in enumerate(sections):
        if len(section) <= chunk_size:
            chunks.append({"text": section.strip(), "source": source, "section": i})
        else:
            # 长section继续按句子分
            sentences = re.split(r'(?<=[。!?.!?])\s*', section)
            current = ""
            for sent in sentences:
                if len(current) + len(sent) <= chunk_size:
                    current += sent
                else:
                    if current:
                        chunks.append({"text": current.strip(), "source": source, "section": i})
                    current = sent
            if current:
                chunks.append({"text": current.strip(), "source": source, "section": i})
    return chunks

# --- API端点 ---
@app.post("/upload")
async def upload_document(file: UploadFile = File(...)):
    """上传文档到知识库"""
    content = await file.read()
    text = content.decode("utf-8")

    chunks = smart_chunk(text, source=file.filename)

    ids = [hashlib.md5(f"{file.filename}_{i}".encode()).hexdigest()[:16]
           for i in range(len(chunks))]
    documents = [c["text"] for c in chunks]
    metadatas = [{"source": c["source"], "section": c["section"]} for c in chunks]

    collection.add(documents=documents, ids=ids, metadatas=metadatas)

    return {"message": f"成功导入 {len(chunks)} 个文档块", "filename": file.filename}

@app.post("/ask", response_model=AnswerResponse)
async def ask_question(req: QuestionRequest):
    """问答接口"""
    # 1. 检索
    results = collection.query(
        query_texts=[req.question],
        n_results=req.top_k,
        where=req.filters
    )

    if not results["documents"][0]:
        raise HTTPException(404, "知识库中没有找到相关信息")

    # 2. 构建上下文
    sources = []
    context_parts = []
    for i, (doc, meta, dist) in enumerate(zip(
        results["documents"][0], results["metadatas"][0], results["distances"][0]
    )):
        context_parts.append(f"[来源{i+1}: {meta.get('source', '未知')}]\n{doc}")
        sources.append({
            "source": meta.get("source", "未知"),
            "content_preview": doc[:100] + "...",
            "relevance": round(1 - dist, 3)
        })

    context = "\n\n---\n\n".join(context_parts)

    # 3. 生成回答
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=2048,
        system="""你是企业知识库AI助手。基于参考资料准确回答问题。
规则:
1. 只使用参考资料中的信息,不编造
2. 用 [来源N] 标注信息出处
3. 如果信息不足,明确告知
4. 回答简洁、专业""",
        messages=[{
            "role": "user",
            "content": f"参考资料:\n{context}\n\n问题:{req.question}"
        }]
    )

    avg_relevance = sum(s["relevance"] for s in sources) / len(sources)

    return AnswerResponse(
        answer=response.content[0].text,
        sources=sources,
        confidence=round(avg_relevance, 2)
    )

@app.get("/stats")
async def get_stats():
    """知识库统计"""
    count = collection.count()
    return {"total_chunks": count}

8.2 项目二:自动代码审查 Agent

Python
# === GitHub PR 自动审查 Agent ===
import anthropic
import subprocess
import json

class CodeReviewAgent:
    """自动代码审查Agent — 集成到CI/CD流水线"""

    def __init__(self):
        self.client = anthropic.Anthropic()
        self.rules = self._load_review_rules()

    def _load_review_rules(self):
        return """
代码审查检查清单:
1. 安全性:SQL注入、XSS、敏感信息泄露、不安全的加密
2. 性能:N+1查询、大循环中的IO、未缓存的重复计算
3. 正确性:边界条件、空指针、并发安全、资源泄漏
4. 可维护性:命名清晰、函数长度、单一职责
5. 测试:新功能是否有测试、测试覆盖边界场景
"""

    def get_pr_diff(self, pr_number=None):
        """获取PR的diff"""
        if pr_number:
            cmd = f"gh pr diff {pr_number}"
        else:
            cmd = "git diff main...HEAD"
        result = subprocess.run(cmd.split(), capture_output=True, text=True)
        return result.stdout

    def review(self, diff: str) -> dict:
        """执行代码审查"""
        response = self.client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=4096,
            system=f"""你是一位资深代码审查专家。

{self.rules}

请以JSON格式返回审查结果:
{{
  "summary": "总体评价",
  "score": 0-10,
  "issues": [
    {{
      "file": "文件路径",
      "line": "行号范围",
      "severity": "critical|warning|info",
      "category": "security|performance|correctness|style",
      "description": "问题描述",
      "suggestion": "修改建议",
      "code_fix": "建议的代码(如适用)"
    }}
  ],
  "highlights": ["做得好的地方"]
}}""",
            messages=[{
                "role": "user",
                "content": f"请审查以下代码变更:\n\n```diff\n{diff[:15000]}\n```"
            }]
        )

        try:
            return json.loads(response.content[0].text)
        except:
            return {"summary": response.content[0].text, "issues": []}

# 使用
# reviewer = CodeReviewAgent()
# diff = reviewer.get_pr_diff(123)
# result = reviewer.review(diff)
# print(json.dumps(result, indent=2, ensure_ascii=False))

8.3 项目三:数据分析对话助手

Python
# === 自然语言 → 数据分析 Agent ===
import anthropic
import pandas as pd
import io
import sys

class DataAnalystAgent:
    """用自然语言分析数据"""

    def __init__(self, df: pd.DataFrame):
        self.client = anthropic.Anthropic()
        self.df = df
        self.schema = self._get_schema()

    def _get_schema(self):
        info = io.StringIO()
        self.df.info(buf=info)
        return f"""数据集信息:
{info.getvalue()}

前5行数据:
{self.df.head().to_string()}

统计摘要:
{self.df.describe().to_string()}
"""

    def analyze(self, question: str) -> dict:
        """根据自然语言问题分析数据"""
        response = self.client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=4096,
            system=f"""你是一位数据分析师。用户会用自然语言提问,你需要编写 Python (pandas) 代码来分析数据。

数据集已加载到变量 `df` 中。

{self.schema}

规则:
1. 输出可执行的Python代码
2. 将分析结果赋值给变量 `result`
3. 如果需要图表,保存为 result_chart.png
4. 代码后附上结果解读""",
            messages=[{"role": "user", "content": question}]
        )

        text = response.content[0].text

        # 提取代码块
        import re
        code_blocks = re.findall(r'```python\n(.*?)```', text, re.DOTALL)

        results = {"explanation": text, "code_outputs": []}

        for code in code_blocks:
            try:
                local_vars = {"df": self.df, "pd": pd}
                exec(code, {"__builtins__": __builtins__}, local_vars)
                if "result" in local_vars:
                    results["code_outputs"].append(str(local_vars["result"]))
            except Exception as e:
                results["code_outputs"].append(f"执行错误: {e}")

        return results

# === 使用示例 ===
# df = pd.read_csv("sales_data.csv")
# agent = DataAnalystAgent(df)
#
# result = agent.analyze("按月份统计销售额趋势,找出增长最快的月份")
# print(result["explanation"])
#
# result = agent.analyze("分析各产品类别的利润率,哪个类别最赚钱?")
# result = agent.analyze("用户地域分布是怎样的?画一个柱状图")

8.4 项目四:多Agent自动化工作流

Python
# === 需求→设计→编码→测试 全自动流水线 ===
import anthropic

client = anthropic.Anthropic()

def call_agent(role: str, system: str, task: str, context: str = "") -> str:
    """统一的Agent调用函数"""
    user_msg = f"任务: {task}"
    if context:
        user_msg = f"上游输出:\n{context}\n\n{user_msg}"

    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=4096,
        system=system,
        messages=[{"role": "user", "content": user_msg}]
    )
    print(f"\n{'='*60}\n[{role}] 完成\n{'='*60}")
    return response.content[0].text

def dev_pipeline(requirement: str):
    """开发流水线: 需求分析 → 架构设计 → 编码 → 测试 → 文档"""

    # Agent 1: 产品经理 — 需求分析
    prd = call_agent(
        "产品经理",
        "你是一位产品经理。将模糊的需求转化为清晰的PRD(产品需求文档)。"
        "包含:功能列表、用户故事、验收标准、优先级排序。",
        f"分析需求并输出PRD: {requirement}"
    )

    # Agent 2: 架构师 — 技术方案
    design = call_agent(
        "架构师",
        "你是一位系统架构师。根据PRD设计技术方案。"
        "包含:技术选型、模块设计、API设计、数据库设计、部署方案。",
        "根据PRD设计技术方案",
        context=prd
    )

    # Agent 3: 开发工程师 — 编码
    code = call_agent(
        "开发工程师",
        "你是一位高级开发工程师。根据技术方案编写完整可运行的代码。"
        "代码必须包含错误处理、日志记录,遵循最佳实践。",
        "根据技术方案编写代码",
        context=design
    )

    # Agent 4: 测试工程师 — 编写测试
    tests = call_agent(
        "测试工程师",
        "你是一位QA工程师。为代码编写全面的测试用例。"
        "包含单元测试、集成测试、边界测试。使用pytest。",
        "为以下代码编写测试用例",
        context=code
    )

    # Agent 5: 技术文档 — 生成文档
    docs = call_agent(
        "技术写作",
        "你是一位技术写作专家。编写清晰的API文档和使用指南。",
        "为以下项目编写技术文档",
        context=f"代码:\n{code}\n\n测试:\n{tests}"
    )

    return {
        "prd": prd,
        "design": design,
        "code": code,
        "tests": tests,
        "docs": docs
    }

# === 执行 ===
# result = dev_pipeline("开发一个URL短链接服务,支持创建短链、跳转、统计点击量")
# 整个流水线约2-3分钟完成,输出完整的PRD+设计+代码+测试+文档
学习路径总结

掌握以上内容后,你具备了以下AI开发能力:

  1. Claude API 开发:对话、工具调用、多模态、流式输出、成本优化
  2. Claude Code 效率:用CLI高效开发,自定义Skill,CLAUDE.md配置
  3. Prompt Engineering:系统级提示设计,Skill封装,推理增强
  4. AI Agent 开发:ReAct、Multi-Agent、记忆系统、生产级架构
  5. RAG 系统:高级检索策略、向量数据库、质量评估
  6. 开源生态:LangChain、LlamaIndex、CrewAI、本地部署
  7. MCP 协议:开发工具服务器,扩展AI能力
  8. 项目实战:知识库、代码审查、数据分析、自动化流水线

下一步:选择一个项目,从零开始构建,在实践中真正掌握这些技能。

Claude & AI Agent 开发实战指南

Claude API · Claude Code · Skill · Agent · RAG · MCP

Built for developers who want to build the future · 2025-2026