LlamaIndex支付集成。
没有 LlamaIndex 包,您不需要一个。将真实的 blockchain0x 客户端包装在 FunctionTool 中,您的代理可以在 Base 上移动 USDC 并门控付费 RAG 查询。
没有 LlamaIndex-specific package,而且你也不需要。LlamaIndex 会用 将普通函数变成工具,因此你只需把真实的 Python client 封装成函数,并将其加入 或 。代理可以发送 USDC、结算 invoice、读取钱包。若要按每次 RAG query 计费,可用同样方式对你自己的 QueryEngineTool 做门控。支付在 Base 上结算。
RAG查询成本高昂。按查询收费使经济合理。
LlamaIndex的优势在于检索增强生成:您拥有一个针对专有文档(您公司的文档、研究数据集、法律语料库)的向量索引,代理通过检索和总结相关块来回答。每个查询在嵌入、LLM令牌和通常的第三方API调用中都需要真实的资金。免费无限访问会消耗预算;按查询计费符合成本结构。
没有为此提供帮助程序,您也不需要一个。您自己进行控制:在 QueryEngineTool 的函数内部,检查调用者是否已付款(您自己的标志,由付款 webhook 切换),然后运行检索或返回价格。钱包功能和控制都是在真正的 blockchain0x 客户端上几行代码,这使得每个查询、每个会话、免费层然后付费的策略完全掌握在您手中。
安装LlamaIndex和核心SDK。一个环境变量。
没有可以添加的 blockchain0x LlamaIndex 包。您安装 LlamaIndex(Python 3.10+,现代的 FunctionAgent / ReActAgent / Workflows API)和真正的 blockchain0x 核心 SDK,然后编写下面的函数。可以与任何向量存储一起使用 - Pinecone、Qdrant、Chroma、Weaviate、LlamaIndex Cloud。
pip install llama-index blockchain0xexport BLOCKCHAIN0X_API_KEY=sk_test_... # sk_test_ = Base Sepolia, sk_live_ = Base mainnet
BLOCKCHAIN0X_API_KEY 是你 dashboard 中的 sk_test_ testnet 或 sk_live_ mainnet key;client 会从环境变量读取它。如果你要对付费查询做门控,webhook handler 还需要 BLOCKCHAIN0X_WEBHOOK_SECRET,该值会在你创建或轮换 webhook 时由 dashboard 仅返回一次。
一个支付的函数,封装为 FunctionTool。
以下是整个集成。send_usdc调用真实的blockchain0x客户端;FunctionTool.from_defaults包装它,FunctionAgent将其作为可调用工具拾取。LlamaIndex读取类型提示和文档字符串以构建架构。将您自己的QueryEngineTool添加到列表中,同一代理也可以回答RAG查询。
from llama_index.core.tools import FunctionTool from llama_index.core.agent.workflow import FunctionAgent from llama_index.llms.openai import OpenAI from blockchain0x import Client blockchain0x = Client() # reads BLOCKCHAIN0X_API_KEY from the environment def send_usdc(agent_id: str, to: str, amount_wei: str) -> str: """Send a USDC payment from an agent wallet. amount_wei is USDC base units (6 decimals), so "10000" is 0.01 USDC. """ return str( blockchain0x.payments.create(body={"agentId": agent_id, "to": to, "amountWei": amount_wei}) ) # Wrap the plain function as a LlamaIndex tool. No dedicated package needed. pay_tool = FunctionTool.from_defaults(fn=send_usdc) agent = FunctionAgent( tools=[pay_tool], # plus your own QueryEngineTool over a RAG index llm=OpenAI(model="gpt-4o"), system_prompt="You pay vendor invoices in USDC within owner-set limits.", ) response = await agent.run( "Pay 0.01 USDC from agent agt_123 to 0xVendor for the dataset." )
When the agent decides to pay, it calls send_usdc, the SDK submits the transfer, and you get a transaction hash back. amount_wei is base units, so 0.01 USDC is "10000". A sk_test_ key keeps it on Base Sepolia until you switch to sk_live_. To charge for a query instead, put the same paid check at the top of your QueryEngineTool's function.
当 webhook 到达时翻转已支付标志。
当支付结算时,Blockchain0x 会将签名的 payment.received 事件 POST 到您的 webhook URL。验证助手包含在 Node SDK 中;在 Python 服务中,您手动根据文档中的 HMAC 进行验证。在您自己的存储中标记呼叫者已付款,您的限制查询工具允许下一个调用通过。下面是 FastAPI 示例。
import hmac, hashlib, os, time from fastapi import FastAPI, Request, HTTPException app = FastAPI() SECRET = os.environ["BLOCKCHAIN0X_WEBHOOK_SECRET"].encode() @app.post("/webhooks/payment") async def receive(request: Request): raw = await request.body() # RAW bytes - do not parse first sig = request.headers.get("X-Blockchain0x-Signature", "") ts = request.headers.get("X-Blockchain0x-Timestamp", "") parts = dict(p.split("=", 1) for p in sig.split(",") if "=" in p) t, v1 = parts.get("t", ts), parts.get("v1", sig) want = hmac.new(SECRET, t.encode() + b"." + raw, hashlib.sha256).hexdigest() if not hmac.compare_digest(want, v1) or abs(time.time() - int(t)) > 300: raise HTTPException(status_code=401) if request.headers.get("X-Blockchain0x-Event-Type") == "payment.received": await trigger_followup() # USDC landed - serve the paid query return {"ok": True}
该算法是针对字符串 t.rawBody 的 HMAC-SHA256,具有常量时间比较和 300 秒的重放窗口。通过 await request.body() 读取原始主体,绝不要 request.json() 重新序列化,因为这会更改签名覆盖的字节。已发布的事件是 payment.received、payment.sent、wallet.deployed 和 webhook.test。对于重的后续工作,排队一个作业(Celery、arq)并立即响应 200,而不是阻塞处理程序。
您正在包装的客户端是开放的。阅读它。
没有可克隆的 LlamaIndex starter package - 上面的 recipe 就是集成方式。blockchain0x SDK 在 GitHub 上开源;这个 recipe 封装了 Python SDK (blockchain0x-python),完整的方法表面已记录在 docs 中。可将其作为 function bodies 的参考。
github.com/tosh-labs/blockchain0x-python完整的 SDK method surface 和 scopes 已记录在 the docs 中。先使用 sk_test_ key 对接 Base Sepolia 进行测试,然后在 function 符合预期后切换到 sk_live_。
五个需要避免的 LlamaIndex 特定陷阱。
这些来自我们的支持收件箱。一旦您知道它们,每个都能节省至少一个小时的调试时间。
没有 LlamaIndex 包 - 您包装 SDK
Blockchain0x 为 LangChain 和 CrewAI 以及 MCP 服务器提供适配器;没有专门的 LlamaIndex 包。上面的配方是路径:一个普通的类型化函数调用真实的 blockchain0x 客户端,使用 FunctionTool.from_defaults 包装。LlamaIndex 读取签名和文档字符串以构建架构,因此请保持文档字符串的准确性。
自己限制QueryEngineTool - 没有助手能做到这一点
付费 RAG 模式是真实且良好的,但没有 paid_query_engine_tool 辅助工具。您构建它:在查询工具的函数中,检查调用者是否已支付(您自己的商店,由下面的 webhook 更新),只有在他们支付的情况下才运行检索,否则返回价格。这只有几行代码,您可以控制门控策略。
金额是USDC基本单位,以字符串形式表示
payments.create takes amountWei: a string of USDC base units, not a float. USDC has 6 decimals, so 0.01 USDC is "10000" and 5 USDC is "5000000". Type the tool argument as str. LlamaIndex passes tool args through pydantic, which will happily coerce a float and lose precision, so keep it a string end to end.
FunctionAgent.run是异步的;SDK调用是同步的
LlamaIndex代理是异步优先的 - await agent.run(...)。您工具中的blockchain0x客户端调用是同步的,这对于单个快速支付是可以的,但在负载下会阻塞循环。如果您正在处理许多并发查询,请使用asyncio.to_thread包装SDK调用,以便事件循环保持响应。
send_payment 可能会提前返回 503
payments.create 默认不重试,并且在链适配器连接到您的网络之前可能返回 503。在您的工具函数内部捕获错误并返回模型可以采取的清晰消息,而不是让代理循环。自动铸造的幂等性密钥意味着手动重试不会重复支付。