当 LangGraph 代理将持久内存与实时搜索基础相结合时，它们会获得巨大的价值。记忆让智能体回忆过去的交互和用户偏好，而搜索接地则通过注入实时数据来防止产生幻觉。本教程构建了一个 LangGraph 代理，该代理在轮流中维护对话状态，使用 Scavio 搜索 API 作为工具，并决定何时进行搜索以及何时依赖内存。该搜索工具每次通话的费用为 0.005 美元，因此即使是健谈的客服人员每月的费用也不会超过几美元。
前置条件
- 已安装 Python 3.10+
- 已安装 langgraph 和 langchain-openai 软件包
- LLM 的 OpenAI API 密钥
- 用于搜索基础的 Scavio API 密钥
操作指南
步骤 1: 定义代理状态模式
LangGraph 使用类型化状态来管理会话流。定义保存消息、搜索结果和内存存储的状态。
Python
from typing import TypedDict, Annotated, Sequence
from langchain_core.messages import BaseMessage
import operator
class AgentState(TypedDict):
messages: Annotated[Sequence[BaseMessage], operator.add]
search_results: str
memory_context: str步骤 2: 构建搜索工具节点
创建一个节点，当代理决定需要当前信息时，该节点会调用 Scavio API。节点接收状态并返回更新的 search_results。
Python
import requests, os
API_KEY = os.environ['SCAVIO_API_KEY']
def search_node(state: AgentState) -> dict:
last_msg = state['messages'][-1].content
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': API_KEY, 'Content-Type': 'application/json'},
json={'query': last_msg, 'country_code': 'us'})
resp.raise_for_status()
results = resp.json().get('organic_results', [])[:5]
context = '\n'.join(f'- {r["title"]}: {r.get("snippet", "")}' for r in results)
return {'search_results': context}步骤 3: 构建内存管理节点
创建一个简单的内存存储来保存对话中的关键事实。代理可以在以后的回合中参考这些内容，而无需重新搜索。
Python
memory_store = {}
def memory_node(state: AgentState) -> dict:
# Extract facts from the latest exchange
messages = state['messages']
if len(messages) >= 2:
user_msg = messages[-2].content if len(messages) >= 2 else ''
ai_msg = messages[-1].content if messages else ''
# Store key topic as memory
key = user_msg[:50]
memory_store[key] = ai_msg[:200]
# Build memory context from recent entries
recent = list(memory_store.items())[-5:]
ctx = '\n'.join(f'Previously discussed: {k} -> {v[:80]}' for k, v in recent)
return {'memory_context': ctx}步骤 4: 构建LLM响应节点
响应节点结合搜索结果、内存上下文和用户消息来生成有根据的、上下文感知的答案。
Python
from langchain_openai import ChatOpenAI
from langchain_core.messages import AIMessage, SystemMessage
llm = ChatOpenAI(model='gpt-4o', temperature=0)
def respond_node(state: AgentState) -> dict:
system = f"""You are a helpful assistant with web search and memory.
Memory from past conversations:
{state.get('memory_context', 'No prior context.')}
Current search results:
{state.get('search_results', 'No search performed.')}
Use search results for factual claims. Use memory to personalize responses."""
messages = [SystemMessage(content=system)] + list(state['messages'])
response = llm.invoke(messages)
return {'messages': [response]}步骤 5: 使用条件路由组装图表
使用路由器将节点连接到 LangGraph，该路由器决定是否直接搜索或回复。有关时事的问题触发搜索；后续操作使用内存。
Python
from langgraph.graph import StateGraph, END
def should_search(state: AgentState) -> str:
last_msg = state['messages'][-1].content.lower()
search_triggers = ['latest', 'current', 'price', '2026', 'today', 'news', 'best']
if any(t in last_msg for t in search_triggers):
return 'search'
return 'respond'
graph = StateGraph(AgentState)
graph.add_node('search', search_node)
graph.add_node('respond', respond_node)
graph.add_node('memory', memory_node)
graph.set_entry_point('search')
graph.add_conditional_edges('search', should_search, {
'search': 'search',
'respond': 'respond'
})
graph.add_edge('search', 'respond')
graph.add_edge('respond', 'memory')
graph.add_edge('memory', END)
app = graph.compile()
print('Agent graph compiled with search + memory nodes')Python 示例
Python
import os, requests
from typing import TypedDict, Annotated, Sequence
from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
import operator
API_KEY = os.environ['SCAVIO_API_KEY']
llm = ChatOpenAI(model='gpt-4o', temperature=0)
class AgentState(TypedDict):
messages: Annotated[Sequence[BaseMessage], operator.add]
search_results: str
def search_node(state):
last = state['messages'][-1].content
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': API_KEY, 'Content-Type': 'application/json'},
json={'query': last, 'country_code': 'us'})
results = resp.json().get('organic_results', [])[:5]
ctx = '\n'.join(f'- {r["title"]}: {r.get("snippet", "")}' for r in results)
return {'search_results': ctx}
def respond_node(state):
sys = f'Use these search results:\n{state.get("search_results", "")}'
msgs = [SystemMessage(content=sys)] + list(state['messages'])
return {'messages': [llm.invoke(msgs)]}
graph = StateGraph(AgentState)
graph.add_node('search', search_node)
graph.add_node('respond', respond_node)
graph.set_entry_point('search')
graph.add_edge('search', 'respond')
graph.add_edge('respond', END)
app = graph.compile()
result = app.invoke({'messages': [HumanMessage(content='Best Python frameworks 2026')], 'search_results': ''})
print(result['messages'][-1].content)JavaScript 示例
JavaScript
// LangGraph is Python-only; this JS example shows the equivalent pattern
const API_KEY = process.env.SCAVIO_API_KEY;
async function searchGrounding(query) {
const resp = await fetch('https://api.scavio.dev/api/v1/search', {
method: 'POST',
headers: { 'x-api-key': API_KEY, 'Content-Type': 'application/json' },
body: JSON.stringify({ query, country_code: 'us' })
});
const data = await resp.json();
return (data.organic_results || []).slice(0, 5)
.map(r => `- ${r.title}: ${r.snippet || ''}`).join('\n');
}
const memory = [];
async function agentTurn(userMessage) {
const context = await searchGrounding(userMessage);
const memoryCtx = memory.slice(-3).join('\n');
const prompt = `Memory:\n${memoryCtx}\n\nSearch:\n${context}\n\nQ: ${userMessage}`;
// Pass prompt to your LLM of choice
console.log(prompt);
memory.push(`User asked: ${userMessage}`);
}
agentTurn('Best Python frameworks 2026').catch(console.error);预期输出
JSON
Agent graph compiled with search + memory nodes
Based on current search results, the best Python frameworks in 2026 are:
1. FastAPI - async-first, ideal for APIs and microservices
2. Django - full-featured for complex web applications
3. Flask - lightweight and flexible for smaller projects
...
Search grounding cost: $0.005 per agent turn