LangGraph 代理可以循环调用工具，这意味着不受约束的搜索工具可能会在单个查询上耗尽信用。本教程构建了一个 LangGraph 代理，将 Scavio 搜索作为工具节点，跟踪每个对话的信用使用情况，并在达到可配置的预算限制时停止搜索。每次搜索费用为 0.005 美元。
前置条件
- Python 3.8+
- 已安装 langgraph 和 langchain
- 来自 scavio.dev 的 Scavio API 密钥
- LLM API 密钥（OpenAI、Anthropic 或本地）
操作指南
步骤 1: 定义预算敏感的搜索工具
创建一个与 LangGraph 兼容的工具来跟踪和限制搜索支出。
Python
import os, requests, json
from langchain_core.tools import tool
from typing import Optional
API_KEY = os.environ['SCAVIO_API_KEY']
SH = {'x-api-key': API_KEY, 'Content-Type': 'application/json'}
search_state = {'calls': 0, 'cost': 0.0, 'max_calls': 10}
@tool
def web_search(query: str, platform: Optional[str] = None) -> str:
"""Search the web for current information. Supports google, reddit, youtube, amazon, walmart platforms."""
if search_state['calls'] >= search_state['max_calls']:
return f"Budget limit reached ({search_state['max_calls']} searches, ${search_state['cost']:.3f}). Synthesize from existing results."
body = {'query': query, 'country_code': 'us'}
if platform: body['platform'] = platform
data = requests.post('https://api.scavio.dev/api/v1/search',
headers=SH, json=body).json()
search_state['calls'] += 1
search_state['cost'] += 0.005
results = data.get('organic_results', [])[:5]
formatted = [f"{r['position']}. {r['title'][:60]} - {r.get('snippet', '')[:100]}" for r in results]
return f"Search results for '{query}' ({search_state['calls']}/{search_state['max_calls']} budget):\n" + '\n'.join(formatted)
print(web_search.invoke({'query': 'best serp api 2026'}))步骤 2: 构建 LangGraph 代理图
使用搜索工具节点和条件路由创建代理图。
Python
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_core.messages import HumanMessage, AIMessage
from typing import TypedDict, Annotated, Sequence
import operator
class AgentState(TypedDict):
messages: Annotated[Sequence, operator.add]
def should_continue(state):
last = state['messages'][-1]
if hasattr(last, 'tool_calls') and last.tool_calls:
return 'tools'
return END
tool_node = ToolNode([web_search])
# Use your preferred LLM:
# from langchain_openai import ChatOpenAI
# llm = ChatOpenAI(model='gpt-4o').bind_tools([web_search])
# Or from langchain_anthropic import ChatAnthropic
# llm = ChatAnthropic(model='claude-sonnet-4-20250514').bind_tools([web_search])
def agent_node(state):
response = llm.invoke(state['messages'])
return {'messages': [response]}
graph = StateGraph(AgentState)
graph.add_node('agent', agent_node)
graph.add_node('tools', tool_node)
graph.set_entry_point('agent')
graph.add_conditional_edges('agent', should_continue, {'tools': 'tools', END: END})
graph.add_edge('tools', 'agent')
app = graph.compile()
print('LangGraph agent compiled with budget-aware search.')步骤 3: 运行具有预算跟踪功能的代理
针对研究问题执行代理并监控信用使用情况。
Python
def run_with_budget(question, max_searches=5):
search_state['calls'] = 0
search_state['cost'] = 0.0
search_state['max_calls'] = max_searches
result = app.invoke({'messages': [HumanMessage(content=question)]})
final = result['messages'][-1].content
print(f'\n--- Agent Response ---')
print(final[:500])
print(f'\n--- Budget ---')
print(f'Searches used: {search_state["calls"]}/{max_searches}')
print(f'Cost: ${search_state["cost"]:.3f}')
return final
# Research question with budget limit
run_with_budget('Compare the top 3 SERP APIs for Python developers in 2026', max_searches=5)步骤 4: 添加多平台搜索策略
配置代理以跨平台搜索更丰富的答案。
Python
def research_with_sources(topic, max_searches=8):
prompt = f"""Research this topic using multiple search platforms for a complete picture:
Topic: {topic}
Strategy:
1. Search Google for overview and top results
2. Search Reddit for real user opinions
3. Search YouTube for tutorial coverage
4. Synthesize findings with sources
Use the platform parameter: google (default), reddit, youtube, amazon
Budget: {max_searches} searches maximum."""
search_state['calls'] = 0
search_state['cost'] = 0.0
search_state['max_calls'] = max_searches
result = app.invoke({'messages': [HumanMessage(content=prompt)]})
print(f'\nResearch complete. {search_state["calls"]} searches, ${search_state["cost"]:.3f}')
return result['messages'][-1].content
research_with_sources('best search API for AI agents', max_searches=8)Python 示例
Python
import os, requests
from langchain_core.tools import tool
SH = {'x-api-key': os.environ['SCAVIO_API_KEY'], 'Content-Type': 'application/json'}
budget = {'n': 0, 'max': 5}
@tool
def web_search(query: str) -> str:
"""Search the web for current information."""
if budget['n'] >= budget['max']:
return 'Budget reached. Use existing results.'
data = requests.post('https://api.scavio.dev/api/v1/search',
headers=SH, json={'query': query, 'country_code': 'us'}).json()
budget['n'] += 1
results = data.get('organic_results', [])[:3]
return '\n'.join(f"{r['title'][:50]}" for r in results)
print(web_search.invoke({'query': 'langgraph tutorial'}))
print(f'Budget: {budget["n"]}/{budget["max"]} (${budget["n"] * 0.005:.3f})')JavaScript 示例
JavaScript
// LangGraph.js equivalent:
const SH = { 'x-api-key': process.env.SCAVIO_API_KEY, 'Content-Type': 'application/json' };
let budget = { n: 0, max: 5 };
async function webSearch(query) {
if (budget.n >= budget.max) return 'Budget reached.';
const data = await fetch('https://api.scavio.dev/api/v1/search', {
method: 'POST', headers: SH,
body: JSON.stringify({ query, country_code: 'us' })
}).then(r => r.json());
budget.n++;
return (data.organic_results || []).slice(0, 3)
.map(r => r.title.slice(0, 50)).join('\n');
}
console.log(await webSearch('langgraph tutorial'));
console.log(`Budget: ${budget.n}/${budget.max} ($${(budget.n * 0.005).toFixed(3)})`);预期输出
JSON
Search results for 'best serp api 2026' (1/5 budget):
1. Scavio - Unified Search API for Developers - Best SERP API with multi-platform...
2. SerpAPI - Google Search API - Reliable Google search results...
3. DataForSEO - SEO Data Provider - Comprehensive SEO data...
LangGraph agent compiled with budget-aware search.
--- Agent Response ---
Based on my research across 3 searches, the top SERP APIs for Python developers...
--- Budget ---
Searches used: 3/5
Cost: $0.015