100 个 Agent Skills 盘点（六）：知识管理与 RAG

1
# Agent 构建简单的 RAG 系统
2
import chromadb
3
from chromadb.utils import embedding_functions
4

5
# 初始化
6
client = chromadb.PersistentClient(path="./chroma_db")
7
ef = embedding_functions.OpenAIEmbeddingFunction(api_key="sk-...")
8
collection = client.get_or_create_collection(
9
    name="documents",
10
    embedding_function=ef
11
)
12

13
# 添加文档
14
collection.add(
15
    documents=["Python 是一种解释型语言", "JavaScript 运行在浏览器中"],
16
    metadatas=[{"source": "wiki"}, {"source": "mdn"}],
17
    ids=["doc1", "doc2"]
18
)
19

20
# 检索
21
results = collection.query(
22
    query_texts=["什么是编程语言？"],
23
    n_results=5,
24
    where={"source": "wiki"}  # 元数据过滤
25
)

1
# Agent 构建生产级 RAG
2
import pinecone
3

4
pinecone.init(api_key="YOUR_API_KEY", environment="us-east-1-aws")
5

6
# 创建索引
7
index = pinecone.Index("knowledge-base")
8

9
# 批量上传向量
10
vectors = [
11
    (f"doc_{i}", embedding, {"text": text, "source": source})
12
    for i, (text, source, embedding) in enumerate(documents)
13
]
14
index.upsert(vectors=vectors, namespace="production")
15

16
# 检索
17
results = index.query(
18
    vector=query_embedding,
19
    top_k=10,
20
    namespace="production",
21
    filter={"source": {"$eq": "official_docs"}},
22
    include_metadata=True
23
)

1
# Agent 使用 Qdrant 构建多模态 RAG
2
from qdrant_client import QdrantClient
3
from qdrant_client.models import Distance, VectorParams, PointStruct
4

5
client = QdrantClient(host="localhost", port=6333)
6

7
# 创建集合（支持多向量）
8
client.create_collection(
9
    collection_name="multimodal",
10
    vectors_config={
11
        "text": VectorParams(size=1536, distance=Distance.COSINE),
12
        "image": VectorParams(size=512, distance=Distance.COSINE)
13
    }
14
)
15

16
# 检索时可以指定使用哪个向量
17
results = client.query_points(
18
    collection_name="multimodal",
19
    using="text",  # 使用文本向量检索
20
    query=text_embedding,
21
    query_filter={
22
        "must": [
23
            {"key": "category", "match": {"value": "technical"}},
24
            {"key": "date", "range": {"gte": "2026-01-01"}}
25
        ]
26
    }
27
)

1
# Agent 使用 Weaviate 混合搜索
2
import weaviate
3

4
client = weaviate.Client("http://localhost:8080")
5

6
# 混合搜索：结合语义和关键词
7
result = (
8
    client.query
9
    .get("Document", ["title", "content", "source"])
10
    .with_hybrid(
11
        query="AI agent 最佳实践",
12
        alpha=0.75  # 0=纯关键词, 1=纯向量
13
    )
14
    .with_where({
15
        "path": ["category"],
16
        "operator": "Equal",
17
        "valueString": "technical"
18
    })
19
    .with_limit(10)
20
    .do()
21
)

1
# Agent 使用不同的 Embedding 模型
2
from openai import OpenAI
3
import requests
4

5
# OpenAI Embedding
6
def get_openai_embedding(text):
7
    client = OpenAI()
8
    response = client.embeddings.create(
9
        model="text-embedding-3-small",
10
        input=text
11
    )
12
    return response.data[0].embedding
13

14
# Jina Embedding（免费、多语言）
15
def get_jina_embedding(text):
16
    response = requests.post(
17
        "https://api.jina.ai/v1/embeddings",
18
        headers={"Authorization": f"Bearer {JINA_API_KEY}"},
19
        json={
20
            "model": "jina-embeddings-v3",
21
            "input": [text]
22
        }
23
    )
24
    return response.json()["data"][0]["embedding"]
25

26
# 本地 BGE 模型（免费、隐私友好）
27
from sentence_transformers import SentenceTransformer
28
model = SentenceTransformer("BAAI/bge-large-zh-v1.5")
29
embedding = model.encode("什么是 RAG？")

1
# Agent 实现多种分块策略
2
from langchain.text_splitter import (
3
    RecursiveCharacterTextSplitter,
4
    MarkdownHeaderTextSplitter
5
)
6

7
# 1. 递归分块（最常用）
8
splitter = RecursiveCharacterTextSplitter(
9
    chunk_size=500,
10
    chunk_overlap=50,
11
    separators=["\n\n", "\n", "。", "！", "？", ".", " "]
12
)
13
chunks = splitter.split_text(document)
14

15
# 2. Markdown 标题分块（保留结构）
16
headers_to_split = [
17
    ("#", "h1"),
18
    ("##", "h2"),
19
    ("###", "h3"),
20
]
21
md_splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers_to_split)
22
chunks = md_splitter.split_text(markdown_document)
23

24
# 3. 语义分块（基于 Embedding 相似度）
25
def semantic_chunk(text, threshold=0.8):
26
    sentences = split_into_sentences(text)
27
    chunks = []
28
    current_chunk = [sentences[0]]
29

30
    for i in range(1, len(sentences)):
31
        similarity = cosine_sim(
32
            get_embedding(sentences[i]),
33
            get_embedding(current_chunk[-1])
34
        )
35
        if similarity > threshold:
36
            current_chunk.append(sentences[i])
37
        else:
38
            chunks.append(" ".join(current_chunk))
39
            current_chunk = [sentences[i]]
40

41
    return chunks

1
# Agent 使用 LlamaIndex 构建 RAG
2
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
3
from llama_index.core.node_parser import SentenceSplitter
4

5
# 加载文档
6
documents = SimpleDirectoryReader("./docs").load_data()
7

8
# 构建索引
9
index = VectorStoreIndex.from_documents(
10
    documents,
11
    transformations=[
12
        SentenceSplitter(chunk_size=512, chunk_overlap=50)
13
    ]
14
)
15

16
# 创建查询引擎
17
query_engine = index.as_query_engine(
18
    similarity_top_k=5,
19
    response_mode="tree_summarize"  # 多文档综合
20
)
21

22
# 查询
23
response = query_engine.query("如何配置 RAG 系统？")
24
print(response)
25
print("Sources:", response.source_nodes)

1
# Agent 使用 LangChain 构建 RAG
2
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
3
from langchain_community.vectorstores import Chroma
4
from langchain.chains import RetrievalQA
5

6
vectorstore = Chroma.from_documents(docs, OpenAIEmbeddings())
7
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})
8

9
qa_chain = RetrievalQA.from_chain_type(
10
    llm=ChatOpenAI(model="gpt-4"),
11
    retriever=retriever,
12
    return_source_documents=True
13
)
14

15
result = qa_chain.invoke({"query": "什么是 Agent Skills？"})

1
# Agent 使用 Mem0 管理记忆
2
from mem0 import Memory
3

4
memory = Memory()
5

6
# 存储记忆
7
memory.add("用户喜欢用 Python 写后端，TypeScript 写前端", user_id="raymin")
8
memory.add("用户的项目使用 Next.js + Supabase 技术栈", user_id="raymin")
9
memory.add("用户偏好简洁的代码风格，不喜欢过度抽象", user_id="raymin")
10

11
# 检索相关记忆
12
relevant = memory.search("帮我写一个新的 API", user_id="raymin")
13
# 返回：用户的语言偏好、技术栈、代码风格偏好
14

15
# 基于记忆个性化回答
16
context = "\n".join([m["text"] for m in relevant])
17
response = llm.generate(
18
    f"基于以下用户偏好回答问题：\n{context}\n\n问题：帮我写一个新的 API"
19
)

1
# Agent 构建技术知识图谱
2
from neo4j import GraphDatabase
3

4
driver = GraphDatabase.driver("bolt://localhost:7687", auth=("neo4j", "password"))
5

6
def create_knowledge_graph(tech_docs):
7
    with driver.session() as session:
8
        for doc in tech_docs:
9
            # 创建技术实体
10
            session.run("""
11
                MERGE (t:Technology {name: $name})
12
                SET t.category = $category,
13
                    t.description = $description
14
            """, name=doc["name"], category=doc["category"], desc=doc["desc"])
15

16
            # 创建依赖关系
17
            for dep in doc["dependencies"]:
18
                session.run("""
19
                    MATCH (a:Technology {name: $from})
20
                    MATCH (b:Technology {name: $to})
21
                    MERGE (a)-[:DEPENDS_ON]->(b)
22
                """, **{"from": doc["name"], "to": dep})
23

24
# 查询：找到从 React 到 PostgreSQL 的依赖路径
25
def find_dependency_path(from_tech, to_tech):
26
    with driver.session() as session:
27
        result = session.run("""
28
            MATCH path = shortestPath(
29
                (a:Technology {name: $from})-[*]-(b:Technology {name: $to})
30
            )
31
            RETURN [n IN nodes(path) | n.name] AS path
32
        """, **{"from": from_tech, "to": to_tech})
33
        return result.single()["path"]

1
# Agent 智能上下文管理
2
class ContextManager:
3
    def __init__(self, max_tokens=8000):
4
        self.max_tokens = max_tokens
5
        self.history = []
6
        self.system_prompt = ""
7

8
    def add_message(self, role, content):
9
        self.history.append({"role": role, "content": content})
10
        self._trim_context()
11

12
    def _trim_context(self):
13
        """智能裁剪上下文"""
14
        total = self._count_tokens()
15

16
        while total > self.max_tokens and len(self.history) > 2:
17
            # 策略1：压缩早期对话
18
            if len(self.history) > 4:
19
                early = self.history[:2]
20
                summary = self._summarize(early)
21
                self.history = [{"role": "system", "content": f"之前的对话摘要：{summary}"}] + self.history[2:]
22

23
            # 策略2：移除低价值消息
24
            if total > self.max_tokens:
25
                self._remove_low_value_messages()
26

27
            total = self._count_tokens()
28

29
    def _remove_low_value_messages(self):
30
        """移除低价值消息（工具调用的中间结果等）"""
31
        for i, msg in enumerate(self.history):
32
            if msg["role"] == "tool" and i > 0:
33
                self.history.pop(i)
34
                break