⚡

AI工程Skill ai-engineer

AI 工程技能专注于构建和部署生产级的大型语言模型（LLM）应用程序、检索增强生成（RAG）系统和 AI 代理。包括 LLM API 集成、向量数据库管理、token 优化等，适用于开发智能聊天机器人、AI 驱动应用等场景。关键词：AI 工程、LLM、RAG、向量数据库、AI 应用开发。

RAG应用 0 次安装 0 次浏览 更新于 3/12/2026

name: ai-engineer description: 构建 LLM 应用程序、RAG 系统和提示管道。用于 LLM 功能、聊天机器人或 AI 驱动的应用程序。

AI 工程

构建生产级 LLM 应用程序和 AI 系统。

何时使用

  • 集成 LLM API
  • 构建 RAG 系统
  • 创建 AI 代理
  • 向量数据库设置
  • Token 优化

LLM 集成

API 设置

from anthropic import Anthropic

client = Anthropic()

def chat(messages: list[dict], system: str = None) -> str:
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        system=system or "You are a helpful assistant.",
        messages=messages
    )
    return response.content[0].text

# 带有重试和错误处理
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=10))
def safe_chat(messages, system=None):
    try:
        return chat(messages, system)
    except Exception as e:
        logger.error(f"LLM 调用失败: {e}")
        raise

结构化输出

import json

def extract_structured(text: str, schema: dict) -> dict:
    prompt = f"""根据以下模式从文本中提取信息：
{json.dumps(schema, indent=2)}

文本: {text}

仅返回有效的 JSON。"""

    response = chat([{"role": "user", "content": prompt}])
    return json.loads(response)

RAG 系统

文档处理

from langchain.text_splitter import RecursiveCharacterTextSplitter

def chunk_documents(docs: list[str], chunk_size=1000, overlap=200):
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=overlap,
        separators=["

", "
", ". ", " "]
    )
    return splitter.split_documents(docs)

向量存储

from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams

client = QdrantClient(":memory:")  # 或 url="http://localhost:6333"

# 创建集合
client.create_collection(
    collection_name="docs",
    vectors_config=VectorParams(size=1536, distance=Distance.COSINE)
)

# 插入向量
client.upsert(
    collection_name="docs",
    points=[
        {"id": i, "vector": embed(chunk), "payload": {"text": chunk}}
        for i, chunk in enumerate(chunks)
    ]
)

# 搜索
results = client.search(
    collection_name="docs",
    query_vector=embed(query),
    limit=5
)

RAG 查询

def rag_query(question: str, top_k=5) -> str:
    # 检索相关块
    results = client.search(
        collection_name="docs",
        query_vector=embed(question),
        limit=top_k
    )

    context = "

".join([r.payload["text"] for r in results])

    prompt = f"""基于以下上下文回答问题。

上下文:
{context}

问题: {question}

回答:"""

    return chat([{"role": "user", "content": prompt}])

成本优化

  • 缓存频繁查询
  • 对简单任务使用较小的模型
  • 尽可能批量请求
  • 跟踪每个功能的 token 使用量
  • 适当设置 max_tokens

示例

输入: “向此应用添加 AI 聊天” 操作: 设置 LLM 客户端，创建聊天端点，添加错误处理

输入: “为文档构建 RAG” 操作: 分块文档，创建嵌入，设置向量存储，实现搜索