手把手教你用LangGraph + GPT-4o-mini + Google Serper API，5分钟搭建一个能联网查资料的智能助手

5分钟打造智能搜索助手LangGraph与GPT-4o-mini实战指南在信息爆炸的时代能够快速获取准确答案的智能助手已成为效率工具的新宠。想象一下当你需要查询实时赛事结果、最新科技动态或生成一张符合特定风格的图片时只需输入简单指令一个融合了语言理解、网络搜索和图像生成能力的AI系统就能自动完成所有工作。本文将带你用LangGraph框架和GPT-4o-mini模型配合Google搜索API构建这样一个全自动化的智能问答系统。1. 环境准备与工具配置工欲善其事必先利其器。在开始构建智能助手前我们需要准备好开发环境和必要的API密钥。这个环节往往是最容易被忽视却最容易踩坑的地方让我们用最稳妥的方式完成基础配置。首先创建一个新的Python虚拟环境推荐3.9版本然后安装核心依赖包pip install langgraph langchain-openai python-dotenv google-search-results接下来是API密钥管理——安全永远是第一要务。在项目根目录创建.env文件存储敏感信息OPENAI_API_KEY你的OpenAI密钥 SERPER_API_KEY你的Google搜索API密钥提示Google Serper API是Google搜索的付费API服务注册后每月有免费额度足够个人开发使用。相比传统爬虫方案它合法稳定且响应结构化。关键工具对象初始化代码如下from dotenv import load_dotenv from langchain_community.tools import GoogleSerperRun from langchain_openai import ChatOpenAI load_dotenv() llm ChatOpenAI(modelgpt-4o-mini, temperature0) search_tool GoogleSerperRun( nameweb_search, description查询网络最新信息适用于时事、体育赛事等实时性强的查询 )2. LangGraph核心架构设计LangGraph的魅力在于它用图结构直观地描述了AI工作流的执行路径。我们的智能助手需要处理两种主要场景直接回答简单问题和调用工具获取实时信息。这正适合用条件边Conditional Edge来实现动态路由。2.1 状态与节点定义首先定义图计算的状态容器——这是各个节点间共享的数据总线from typing import TypedDict, Annotated from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] # 消息历史记录然后实现两个核心节点聊天节点调用LLM处理用户输入工具节点执行搜索等外部操作def chat_node(state: AgentState): # 绑定可用工具 llm_with_tools llm.bind_tools([search_tool]) # 处理最新消息 response llm_with_tools.invoke(state[messages]) return {messages: [response]} def tool_node(state: AgentState): last_msg state[messages][-1] # 提取工具调用指令 tool_calls getattr(last_msg, tool_calls, []) # 执行搜索 search_result search_tool.invoke(tool_calls[0][args]) return {messages: [ToolMessage( contentstr(search_result), tool_call_idtool_calls[0][id] )]}2.2 条件路由与图装配智能决策的核心在于路由函数它决定下一步执行哪个节点from langgraph.graph import END def router(state: AgentState): last_msg state[messages][-1] if hasattr(last_msg, tool_calls): return tool_node # 需要执行工具 return END # 直接结束现在将这些组件装配成完整的工作流from langgraph.graph import StateGraph builder StateGraph(AgentState) builder.add_node(chat, chat_node) builder.add_node(tool, tool_node) builder.set_entry_point(chat) # 条件边配置 builder.add_conditional_edges( chat, router, {tool_node: tool, END: END} ) builder.add_edge(tool, chat) flow builder.compile()3. 实战演示与效果优化让我们用实际案例验证这个智能助手的表现。查询实时体育赛事是个很好的测试场景因为这类信息无法仅靠LLM的内部知识回答。3.1 基础查询测试result flow.invoke({ messages: [(user, 2024年法国网球公开赛男单冠军是谁)] }) for msg in result[messages]: print(f{msg.type}: {msg.content})典型输出流程user: 原始问题ai: 生成工具调用指令tool: 返回搜索结果ai: 格式化最终答案3.2 性能优化技巧通过几个简单调整可以显著提升助手表现提示词工程from langchain_core.prompts import ChatPromptTemplate prompt ChatPromptTemplate.from_messages([ (system, 你是一个专业的信息助手。 当问题涉及实时信息时请使用搜索工具。 回答要简明扼要不超过3句话。), (user, {input}) ])结果后处理def format_results(raw): # 移除广告等无关内容 return \n.join(raw.split(\n)[:3])错误处理增强def safe_search(query): try: return search_tool.invoke(query) except Exception: return 当前无法获取信息请稍后再试4. 进阶功能扩展基础版助手已经能处理简单查询但要让其真正实用还需要一些进阶功能。4.1 多工具集成除了搜索我们可以轻松集成其他工具比如from langchain_community.tools import OpenAIDALLEImageGenerationTool dalle_tool OpenAIDALLEImageGenerationTool( nameimage_gen, description根据文字描述生成图片 ) # 更新工具列表和路由逻辑 tools [search_tool, dalle_tool] llm_with_tools llm.bind_tools(tools) def enhanced_router(state): last_msg state[messages][-1] if not hasattr(last_msg, tool_calls): return END tool_name last_msg.tool_calls[0][name] return { web_search: search_node, image_gen: image_node }.get(tool_name, END)4.2 记忆与上下文添加对话记忆能让助手更智能from langchain.memory import ConversationBufferMemory memory ConversationBufferMemory( return_messagesTrue, memory_keyhistory ) def chat_with_memory(state): # 合并历史消息 full_context memory.load_memory_variables({})[history] state[messages] response llm_with_tools.invoke(full_context) memory.save_context( {input: state[messages][-1].content}, {output: response.content} ) return {messages: [response]}4.3 流式输出对于长时间操作流式响应能极大提升用户体验from langchain_core.runnables import RunnableLambda async def stream_response(query): async for chunk in flow.astream( {messages: [(user, query)]} ): if messages in chunk: yield chunk[messages][-1].content5. 生产环境部署建议当开发完成后如何将这个小助手变成真正的服务以下是关键考量点API封装from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class Query(BaseModel): text: str app.post(/ask) async def ask(query: Query): result flow.invoke({messages: [(user, query.text)]}) return {answer: result[messages][-1].content}性能监控import time from collections import defaultdict stats defaultdict(list) def with_metrics(func): def wrapper(*args, **kwargs): start time.time() try: result func(*args, **kwargs) stats[success].append(time.time()-start) return result except Exception: stats[error].append(time.time()-start) raise return wrapper限流保护from fastapi import HTTPException from slowapi import Limiter from slowapi.util import get_remote_address limiter Limiter(key_funcget_remote_address) app.state.limiter limiter app.post(/ask) limiter.limit(10/minute) async def ask(query: Query, request): ...在实际项目中我发现最影响用户体验的往往是异常处理——当搜索引擎返回意外结果或API超时时如何优雅降级比追求完美响应更重要。一个实用的技巧是为每种常见错误类型预设友好的回复模板这比直接显示技术错误信息要好得多。