CLAP - Cognitive Layer Agent Package

CLAP - 认知层 Agent 包

CLAP (Cognitive Layer Agent Package) 是一个 Python 框架，为基于现代 Agent 模式创建精复杂 AI Agent 提供了基础构件。它使开发者能够轻松构建具有推理、规划和与外部工具、系统和知识库交互能力的 Agent。

采用异步核心（asyncio）构建，CLAP 为复杂的 Agent 工作流提供灵活性和性能。

## 主要功能

• 模块化 Agent 模式：
  • ReAct Agent： 实现 Reason-Act 循环，具有强大的思维提示和原生工具调用。适合复杂推理和 RAG。
  • Tool Agent： 一个简单直接的 Agent，用于单步工具使用，包括简单 RAG。
  • 多 Agent 团队： 定义具有依赖关系的专业化 Agent 团队，支持协作任务执行（顺序或并行）。
• 高级工具集成：
  • 原生大模型工具调用： 利用现代大模型 API 实现可靠的工具执行。
  • 本地工具： 使用 @tool 装饰器轻松定义和使用本地 Python 函数（同步和异步均支持）作为工具。
  • 远程工具 (MCP)： 通过包含的 MCPClientManager 与 Model Context Protocol (MCP) 服务集成，允许 Agent 发现和使用外部系统暴露的工具（目前支持 SSE 传输）。
  • 强大的验证和类型强制转换： 使用 jsonschema 严格验证工具参数，并针对常见大模型输出尝试类型强制转换（例如，字符串数字转为整数）。
• 检索增强生成 (RAG) 能力：
  • VectorStoreInterface： 与各种向量数据库交互的抽象。
  • 支持的向量存储：
    • ChromaDB： (ChromaStore) 用于本地或可自托管向量存储。
    • Qdrant： (QdrantStore) 用于本地（内存或基于文件）向量存储。
  • EmbeddingFunctionInterface： 与不同嵌入模型一致交互的协议。
  • 支持的嵌入函数包装器：
    • SentenceTransformerEmbeddings：使用 sentence-transformers 库中的模型。
    • OllamaEmbeddings：使用通过 Ollama 在本地运行的模型生成嵌入。
    • FastEmbedEmbeddings：利用 fastembed 库进行 CPU 优化的嵌入。（注：通过异步包装器进行大规模批量摄取的性能可能因 CPU 而异，初始批量加载时可能比 SentenceTransformers 更慢。）
  • RAG 感知 Agent： Agent（通过 ReactAgent）和 ToolAgent 都可以配备 vector_store 来执行 vector_query 工具调用，使其能够在响应前检索上下文。
  • 工具函数： 在 clap.utils.rag_utils 中包含基础 PDF 和 CSV 文本加载器以及分块策略。
• 可插拔大模型后端：
  • 使用策略模式（LLMServiceInterface）来抽象大模型交互。
  • 包含现成的服务实现：
    • Groq： (GroqService)
    • Google Generative AI (Gemini)： (GoogleOpenAICompatService 通过 OpenAI 兼容层)
    • Ollama (本地大模型)： (OllamaOpenAICompatService 也称为 OllamaService 通过 OpenAI 兼容层，支持使用本地运行的模型如 Llama 3、Mistral 等)
  • 易于扩展以支持其他大模型提供商。
• 异步核心： 完全基于 asyncio 构建，用于高效 I/O 操作和潜在的并发。
• 结构化上下文传递： 实现团队中 Agent 之间的清晰有序信息流。
• 内置工具： 包含网络搜索的辅助工具 (duckduckgo_search)。更多工具可通过可选依赖获取。

安装

确保已安装 Python 3.10 或更高版本。

pip install clap-agents

确保已安装 Python 3.10 或更高版本。

pip install clap-agents


To use specific features, you might need to install optional dependencies:
# For Qdrant support (includes fastembed)
pip install "clap-agents[qdrant]"

# For ChromaDB support
pip install "clap-agents[chromadb]"

# For Ollama (LLM and/or Embeddings)
pip install "clap-agents[ollama]"

# For other tools like web crawling or visualization
pip install "clap-agents[standard_tools,viz]"

# To install all major optional dependencies
pip install "clap-agents[all]"

检查 pyproject.toml 以查看完整的 [project.optional-dependencies] 列表。如果使用本地 Ollama 或 Qdrant，还需要运行这些外部服务。 根据你打算使用的工具或大模型后端，你可能需要 pyproject.toml 中列出的额外依赖（例如 groq、openai、mcp、jsonschema、requests、duckduckgo-search、graphviz）。检查 [project.dependencies] 和 [project.optional-dependencies] 部分。

快速入门：使用本地工具的简单工具调用 Agent

此示例展示如何使用 Groq 后端和本地工具创建工具调用 Agent。

from dotenv import load_dotenv
from clap import ToolAgent
from clap import duckduckgo_search

load_dotenv()

async def main():
    agent = ToolAgent(tools=duckduckgo_search, model="meta-llama/llama-4-scout-17b-16e-instruct")
    user_query = "Search the web for recent news about AI advancements."
    response = await agent.run(user_msg=user_query)
    print(f"Response:\n{response}")

asyncio.run(main())

快速入门：使用本地工具的简单 ReAct Agent

此示例展示如何使用 Groq 后端和本地工具创建 ReAct Agent。

import asyncio
import os
from dotenv import load_dotenv
from clap import ReactAgent, tool, GroqService

load_dotenv() 
@tool
def get_word_length(word: str) -> int:
    """Calculates the length of a word."""
    print(f"[Local Tool] Calculating length of: {word}")
    return len(word)

async def main():
    groq_service = GroqService() # Your service of choice (either groq or Google)
    agent = ReactAgent(
        llm_service=groq_service,
        model="llama-3.3-70b-versatile", # Or another Groq model
        tools=[get_word_length], # Provide the local tool
        # system_prompt="You are a helpful assistant." # Optional base prompt
    )

    user_query = "How many letters are in the word 'framework'?"
    response = await agent.run(user_msg=user_query)
    
    print(response)
    
asyncio.run(main())

快速入门：使用 Ollama 的简单工具调用 Agent

此示例展示如何使用本地 Ollama 模型和本地工具创建 ToolAgent。 请确保 Ollama 正在运行，并且已拉取模型（例如，运行 ollama pull llama3）。

import asyncio
from dotenv import load_dotenv
from clap import ToolAgent, tool, OllamaService # Assuming OllamaService is your OllamaOpenAICompatService

load_dotenv()

@tool
def get_capital(country: str) -> str:
    """Returns the capital of a country."""
    if country.lower() == "france": return "Paris"
    return f"I don't know the capital of {country}."

async def main():
    # Initialize the Ollama service
    ollama_llm_service = OllamaService(default_model="llama3") # Specify your Ollama model

    agent = ToolAgent(
        llm_service=ollama_llm_service,
        model="llama3", # Model name for this agent
        tools=[get_capital]
    )
    user_query = "What is the capital of France?"
    response = await agent.run(user_msg=user_query)
    print(f"Query: {user_query}\nResponse:\n{response}")

    await ollama_llm_service.close() # Important for OllamaService

if __name__ == "__main__":
    asyncio.run(main())

快速入门：使用 Qdrant 和 Ollama 嵌入的 RAG Agent

此示例展示 Agent 如何使用 Ollama 生成嵌入，并以 Qdrant 作为向量存储来执行 RAG。 请确保 Ollama 正在运行（已拉取 nomic-embed-text 和 llama3），并且 Qdrant 正在运行（例如，通过 Docker 运行）。

import asyncio
import os
import shutil
from dotenv import load_dotenv
from clap import Agent, QdrantStore, OllamaEmbeddings, OllamaService
from clap.utils.rag_utils import chunk_text_by_fixed_size
from qdrant_client import models as qdrant_models # If needed for distance

load_dotenv()

OLLAMA_HOST = "http://localhost:11434"
EMBED_MODEL = "nomic-embed-text"
LLM_MODEL = "llama3"
DB_PATH = "./temp_rag_db_ollama_qdrant"
COLLECTION = "my_rag_docs"

async def main():
    if os.path.exists(DB_PATH): shutil.rmtree(DB_PATH)

    ollama_ef = OllamaEmbeddings(model_name=EMBED_MODEL, ollama_host=OLLAMA_HOST)
    vector_store = await QdrantStore.create(
        collection_name=COLLECTION,
        embedding_function=ollama_ef,
        path=DB_PATH, # For local file-based Qdrant
        recreate_collection_if_exists=True
    )

    sample_texts = ["The sky is blue due to Rayleigh scattering.", "Large language models are powerful."]
    chunks = [chunk for text in sample_texts for chunk in chunk_text_by_fixed_size(text, 100, 10)]
    ids = [str(i) for i in range(len(chunks))] # Qdrant needs UUIDs; QdrantStore handles this
    
    if chunks:
        await vector_store.add_documents(documents=chunks, ids=ids)
    print(f"Ingested {len(chunks)} chunks.")

    ollama_llm_service = OllamaService(default_model=LLM_MODEL, base_url=f"{OLLAMA_HOST}/v1")
    rag_agent = Agent(
        name="RAGMaster",
        backstory="I answer questions using provided documents.",
        task_description="Why is the sky blue according to the documents?", # This becomes the User Query
        llm_service=ollama_llm_service,
        model=LLM_MODEL,
        vector_store=vector_store
    )

    response = await rag_agent.run()
    print(f"Query: {rag_agent.task_description}\nResponse:\n{response.get('output')}")

    await vector_store.close()
    await ollama_llm_service.close()
    if os.path.exists(DB_PATH): shutil.rmtree(DB_PATH)

asyncio.run(main())

v0.3.0 新增：Web3 和链上 Agent 功能

CLAP 现在包含一套强大的工具包，可用于构建能够直接与 Ethereum 等兼容 EVM 的区块链交互的自主 Agent。你的 Agent 现在可以持有资产、执行交易并与智能合约交互，为 DeFi、DAO 和链上自动化开启全新的可能性。 设置 要启用 Web3 功能，请安装 web3 extra：

pip install "clap-agents[web3]"

你还需要在 .env 文件中设置以下变量：

# Your connection to the blockchain (e.g., from Alchemy or Infura)
WEB3_PROVIDER_URL="https://sepolia.infura.io/v3/YOUR_API_KEY"

# The private key for your agent's wallet.
# WARNING: For testing only. Do not use a key with real funds.
AGENT_PRIVATE_KEY="0xYourTestnetPrivateKeyHere"

核心 Web3 工具

该框架现在包含一系列预先构建的可靠工具，用于链上交互：

get_erc20_balance：检查钱包中任意标准 ERC-20 代币的余额。

wrap_eth：将原生 ETH 转换为 WETH（Wrapped Ether），这是与许多 DeFi 协议交互的必要步骤。

swap_exact_tokens_for_tokens：在 Uniswap V3 上执行交易，让你的 Agent 能够自主重新平衡其投资组合。

get_token_price：从链上 Chainlink 预言机获取实时资产价格，支持基于数据做出决策。

interact_with_contract：一种功能强大的通用工具，可根据地址和 ABI 调用任意智能合约中的任意函数。

快速入门：简单的 DeFi Agent

此示例展示一个 Agent 如何将 ETH 包装为 WETH，然后将其兑换为另一种代币，这是常见的 DeFi 任务。

import os
import asyncio
from dotenv import load_dotenv
from clap import ReactAgent, GroqService
from clap.tools import wrap_eth, swap_exact_tokens_for_tokens

load_dotenv()

# --- Configuration ---
WETH_ADDRESS = "0xfFf9976782d46CC05630D1f6eBAb18b2324d6B14" # WETH on Sepolia
USDC_ADDRESS = "0x1c7D4B196Cb0C7B01d743Fbc6116a902379C7a98" # USDC on Sepolia

async def main():
    # We use a ReactAgent for multi-step reasoning
    agent = ReactAgent(
        llm_service=GroqService(),
        tools=[wrap_eth, swap_exact_tokens_for_tokens],
        model="llama-3.3-70b-versatile",
        system_prompt="You are a DeFi agent. You execute financial transactions precisely as instructed.",
        # For on-chain tasks, sequential execution is safer to avoid race conditions
        parallel_tool_calls=False 
    )

    # A clear, two-step task for the agent
    user_query = f"""
    First, wrap 0.01 ETH.
    Second, after the wrap is successful, swap that 0.01 WETH for USDC.
    The WETH address is {WETH_ADDRESS} and the USDC address is {USDC_ADDRESS}.
    """

    print("--- Running Simple DeFi Agent ---")
    response = await agent.run(user_msg=user_query, max_rounds=5)
    
    print("\n--- Agent Final Response ---")
    print(response)

if __name__ == "__main__":
    asyncio.run(main())

这项新功能将你的 CLAP Agent 从单纯的观察者转变为去中心化经济中的积极参与者。

进一步探索

多 Agent 团队：查看 examples/test_clap_comprehensive_suite.py 和其他团队示例，了解如何设置顺序或并行的 Agent 工作流。

MCP 集成：查看 examples/test_clap_comprehensive_suite.py（请确保 examples/simple_mcp.py 等示例中的相应 MCP 服务正在运行）。

其他大模型服务（Groq、Google Gemini、Ollama）：修改快速入门示例以使用 GroqService 或 GoogleOpenAICompatService（请确保已设置 API 密钥）。

不同的向量存储和嵌入函数：如综合测试套件所示，尝试使用 ChromaStore、QdrantStore、SentenceTransformerEmbeddings、FastEmbedEmbeddings 和 OllamaEmbeddings。

许可证 本项目采用 Apache License 2.0 条款授权。详情请参阅 LICENSE 文件。