Claude & AI Agent 开发实战指南

0.1 什么是大语言模型（LLM）

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

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

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

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

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

Transformer 架构：LLM 的心脏

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

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

0.2 Token — AI 的基本计量单位

什么是 Token？

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

Token 的重要性

如何估算 Token 数

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

上下文窗口（Context Window）

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

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

参数是什么？

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

主流模型参数规模参考

量化（Quantization）：缩小模型大小的技巧

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

0.4 核心概念速查

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

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

Embedding（向量嵌入） — 把文字变成数字

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

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

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

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

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

Hallucination（幻觉）— LLM 最大的缺陷

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

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

主要 AI 提供商与旗舰模型（2025-2026）

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

成本计算实例

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

0.6 如何选择适合的模型

决策树：选模型不再迷茫

按使用场景推荐

1.1 API 快速入门 API

Claude 模型家族概览（2025-2026）

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}")

1.3 Tool Use / Function Calling 核心

让 Claude 调用你的代码

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

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

配置文件层级

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 的快捷操作。在会话中输入 / 即可看到可用命令。

自定义 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. 用以下格式输出：

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

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

## 遇到的问题
- {如果有未解决的问题}

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/ 下的已有迁移文件

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\"'"
      }
    ]
  }
}

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.1 Agent 架构设计 核心

Agent 的核心组件

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 交替进行"思考"和"行动"。

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 框架对比 选型

5.1 高级检索策略 核心

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 向量数据库选型

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.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本地部署

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.1 MCP 概述 新标准

什么是 MCP？

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

MCP 的三大能力

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.1 项目一：企业知识库问答系统 完整项目

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+设计+代码+测试+文档