Langgraph学习4：搭一个web搜索agent

在前面的Langgraph学习文章中，简单搭了一个Langgraph的基础流程和如何让LLM调用工具。本篇继续深入，将结合基本chatbot流程，搭建一个能够进行Web搜索的AI助手。

1. 集成搜索API

首先我们需要一个搜索工具。可以直接使用Langchain提供的Tavily搜索API，它是一个针对AI优化的搜索引擎，不过它的官方个人免费账户每个月只能调用1000次，这里我们先选择免费的的DuckDuckGo搜索：

from langchain_community.tools.ddg_search.tool import DuckDuckGoSearchResults
from langchain_core.tools import tool

@tool
def search(query: str):
    """用于浏览网络进行搜索。"""
    # 可选的搜索工具: TavilySearchResults(max_results=3)
    search_tool = DuckDuckGoSearchResults(num_results=3) #这个配置根据需求来设置 output_format="list"
    return search_tool.invoke(query)

这里我们使用DuckDuckGoSearchResults作为搜索引擎，并设置最大返回结果数为3，输出格式为列表。注意DuckDuckGoSearchResults要使用num_results

2. 自定义搜索工具节点

与第二篇文章不同，这次不使用Langgraph的ToolNode，而是自定义一个更灵活的搜索节点：

tools_by_name = {tool.name: tool for tool in tools}

def search_tool_node(state: dict):
    result = []
    for tool_call in state["messages"][-1].tool_calls:
        tool = tools_by_name[tool_call["name"]]
        observation = tool.invoke(tool_call["args"])
        print('工具搜索结果的完整输出------>',observation,'\n')
        # 直接转为字符串
        search_result = str(observation)
        result.append(ToolMessage(content=search_result, tool_call_id=tool_call["id"]))
    return {"messages": result}

这个节点做了几件事：

从状态中获取最后一条消息的工具调用信息
根据工具名称找到对应的工具函数
执行工具调用并获取结果
将结果转换为字符串（这一步比较关键！）
将结果封装为ToolMessage并返回

自定义节点比使用预设的ToolNode更灵活，方便我们添加更多处理逻辑。特别是第4步，将搜索结果转换为字符串是解决与LLM兼容性问题的关键。

处理搜索结果格式问题

在实际测试中，如果直接将搜索工具返回的原始对象传递给LLM，会遇到以下错误：

graph.astream_events执行出错: Error code: 400 - {'code': 20029, 'message': 'Only text and image_url are supported.', 'data': None}

这是因为大多数LLM API只支持文本和图片URL作为输入，而TavilySearchResults搜索工具返回的是一个复杂的Python对象（通常是列表或字典），而DuckDuckGoSearchResults如果不加output_format=”list”的配置天然返回的就是个字符串。所以通过简单地使用str(observation)将结果转换为字符串，可以解决这个问题，确保了与LLM API的兼容性。如果是需要手动处理好格式，比如对list做一个处理，就在这里直接把处理完的结构再改为字符串输出给LLM。

3. 条件路由函数

条件路由函数决定了工作流的执行路径：

def route_tools(state: MessagesState):
    """
    在条件边中使用,如果最后一条消息包含工具调用,则路由到工具节点,否则路由到结束节点。
    """
    messages = state['messages']
    last_message = messages[-1]
    # 检查是否是AI消息且有工具调用
    if hasattr(last_message, 'tool_calls') and last_message.tool_calls:
        return "search_tool"  # 如果有工具调用，路由到工具节点
    
    return END

4. 构建工作流

接下来完成工作流的构建：

# 创建图构建器
graph_builder = StateGraph(MessagesState)

# 添加节点
graph_builder.add_node("chatbot", chatbot_stream)
graph_builder.add_node("search_tool", search_tool_node)

# 设置入口点
graph_builder.set_entry_point("chatbot")

# 添加条件边
graph_builder.add_conditional_edges(
    "chatbot",
    route_tools
)

# 添加从搜索工具节点返回到聊天节点
graph_builder.add_edge("search_tool", "chatbot")

# 编译图
graph = graph_builder.compile()

这个工作流的执行路径有两种可能：

chatbot → END （如果LLM直接回答，不需要搜索）
chatbot → search_tool → chatbot → END （如果需要搜索信息）

第二种路径中，search_tool执行完后会回到chatbot，让LLM基于搜索结果生成最终回复。这是一个循环结构，允许LLM在必要时多次调用搜索工具。

5. 定义LLM处理节点

我们需要一个节点来处理用户输入并生成LLM响应：

async def chatbot_stream(state: MessagesState):
    """生成流式回复的节点函数"""
    print('chatbot_stream收到的完整输入------>',state,'\n')
    messages = state["messages"]
    
    # 使用非流式方式接收完整返回
    response = llm_with_tools.invoke(
        messages,
        functions=functions,
        function_call="auto"
    )
    return {"messages": [response]}

这个函数接收当前的消息历史作为输入，使用LLM生成回复。我们设置function_call="auto"，让LLM自行决定是否需要调用工具。

6. 准备System Prompt和测试

为了引导LLM正确使用搜索工具，这里设计了一个详细的System Prompt：

today = datetime.now().strftime("%Y-%m-%d")
system_message = SystemMessage(content=f"""
    # 你是一个善于分析的AI助手。
    ## 对于用户的所有问题，先分析是否需要调用搜索工具，再进行回复，比如查询东西，需要调用搜索工具，如果只是问问题，则不需要调用搜索工具。
    ## 请牢记今天的日期是{today},调用工具时，直接使用{today}的日期。
    - 分析示例:比如用户询问日历,黄历,新闻等实时查询相关的需要调用工具;比如询问今天的新闻或今天的日历。直接把{today}的日期作为参数传给搜索工具去做搜索。
    - 每个搜索结果都需要按照以下格式返回：
        - 标题：
        - 摘要：
        - 链接：
""")

System Prompt的设计包含了几个关键要素：

明确角色定位：一个善于分析的AI助手
决策指导：什么情况下应该调用搜索工具
时间意识：使用当前日期进行搜索
输出格式：要求以特定格式呈现搜索结果

测试问题则是关于室内适合种植的花卉：

first_message = HumanMessage(content="""
    请帮我搜索下室内适合种植什么花?
""")

7. 执行流程与事件监听

在执行过程中，我们使用astream_events方法捕获工作流中的各种事件：

async for event in graph.astream_events(initial_state, config={"configurable": {"thread_id": "1"}}, version="v2"):
    event_type = event['event']
    if event_type == 'on_tool_start' and event['data']:
        print('开始调用工具查询', event['data'],'\n\n')
    elif event_type == 'on_tool_end' and event['data']:
        print('工具查询结束',event['data'],'\n\n')
    elif event_type == 'on_chat_model_stream':
        print('on_chat_model_stream事件------>',event["data"]["chunk"].content,'\n\n')
        on_chat_model_stream_list.append(event["data"]["chunk"].content)

通过监听这些事件，我们可以跟踪整个工作流的执行过程：

on_tool_start：工具开始执行时触发
on_tool_end：工具执行完成时触发
on_chat_model_stream：LLM生成输出时触发

这样我们可以实时观察到整个交互的流程。

8. 测试结果示例

问题：请帮我搜索下室内适合种植什么花?

以下是示例输出：

生成完成！ 为了给您提供合适的建议，我将搜索一些适合室内种植的花卉。


根据搜索结果，这里有一些适合室内种植的花卉推荐：

1. **长寿花**
   - 摘要：这是小美最喜欢的室内盆栽花卉，适合室内种植。
   - 链接：[可以放室内养的花有哪些（100种常见室内盆栽花卉图鉴）](https://www.huaguozhijia.com/yanghua/13493.html)

2. **水仙花**
   - 摘要：水仙花是一种时兴的室内水养植物，球茎运用简易，适合放在桌面或阳台上。
   - 链接：[适合室内养绿植推荐：花卉绿植盆栽、高级感、大型绿植、水培、旺宅、养不死的室内绿植 - 知乎](https://zhuanlan.zhihu.com/p/21133659555)

3. **十大室内花卉排名**
   - 摘要：在忙碌的都市生活中，越来越多的人喜欢在室内种植花卉，以增添生活情趣，净化空气。本文介绍了十大室内花卉排名，以及怎样挑选和养护这些美丽的室内花卉。
   - 链接：[十大室内花卉排名；哪些花卉最适合室内种植；怎样挑选和养护花卉](https://baijiahao.baidu.com/s?id=1812440518891264633)

4. **10种适合在室内养护的花卉**
   - 摘要：经过十年的养花经验，作者向大家介绍10种非常适合室内种植的花卉。
   - 链接：[10种适合在室内养护的花卉 - 月季网](https://www.yuejiw.com/post/29176.html)

您可以根据自己的喜好和养护条件选择合适的花卉进行种植。

问题：最近AI领域有什么新闻?

以下是示例输出：

根据搜索结果，以下是最近AI领域的几条新闻：

1. 标题：科技资讯ai速递：昨夜今晨科技热点一览 丨2025年4月11日
   摘要：2025年04月11日 06:24 新浪ai. 缩小 ... 板 成为全网最火热的ai硬件 2025年爆火的"小智ai"开源项目，凭借其拟人化语音交互和低门槛特性，两个月内接入 ...
   链接：[科技资讯ai速递：昨夜今晨科技热点一览 丨2025年4月11日](https://news.sina.com.cn/zx/ds/2025-04-11/doc-inestvsi2877509.shtml)

2. 标题：Ai日报 - 2025年4月11日
   摘要：📊 趋势图谱：未来6个月，ai在个性化营销、自动化客户服务、供应链优化方面的应用将持续深化；生成式ai在产品设计、广告创意领域的应用将增加。 3.3 金融服务 ai. 🌐 全球视角：ai在金融领域的应用广泛，但也面临严格监管和潜在风险。
   链接：[Ai日报 - 2025年4月11日 - 腾讯云](https://cloud.tencent.com/developer/article/2512975)

3. 标题：中国学者领衔研发AI肿瘤预测模型
   摘要：2025-04-11 08:30 发布于北京 人民网资讯精选官方账号 本报电(申奇)近期，由斯坦福大学医学院癌症研究所主导、中国学者领衔进行的一项研究发表于 ...
   链接：[中国学者领衔研发AI肿瘤预测模型_腾讯新闻] (https://news.qq.com/rain/a/20250411A020P200)

这个回复是在LLM接收到由字符串形式的搜索结果后生成的。如果没有将搜索结果转换为字符串，就会遇到前面提到的错误。

总结

在这篇文章中，我们成功构建了一个能够进行网络搜索的AI助手。相比前面的简单工具示例，这个助手可以搜索互联网上的实时信息，使其回答更加准确和有用。

关键改进点包括：

集成了实际可用的DuckDuckGo搜索API
处理了工具返回结果格式兼容性问题

graph图例流程和上一篇没什么变化，只是工具名称改为了一个具象的工具名：

在下一篇中，我们将进一步扩展这个助手，让它不仅能搜索，还能抓取和分析网页内容，以提供更深入的信息处理能力。

完整代码：

from datetime import datetime
import os
from dotenv import load_dotenv
load_dotenv()
import asyncio
from langchain_openai import ChatOpenAI
from langchain_community.chat_models import QianfanChatEndpoint
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_community.tools.ddg_search.tool import DuckDuckGoSearchResults
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, SystemMessage,ToolMessage
from langgraph.graph import StateGraph, START, END,MessagesState
from langchain_core.utils.function_calling import convert_to_openai_function
from langchain_core.runnables.graph import MermaidDrawMethod
from langchain_core.messages import ToolMessage
from langchain_core.tools import tool

# 创建图构建器
graph_builder = StateGraph(MessagesState)

os.environ['TAVILY_API_KEY'] = os.getenv('TAVILY_API_KEY', '')

# 创建工具
@tool
def search(query: str):
    """用于浏览网络进行搜索。"""
    # search_tool = TavilySearchResults(max_results=3)
    search_tool = DuckDuckGoSearchResults(num_results=3, output_format="list") # output_format="list"
    return search_tool.invoke(query)
tools = [search]

# llm = QianfanChatEndpoint(
#     model="ernie-lite-pro-128k",
#     api_key=os.getenv('QIANFAN_AK', ''),
#     secret_key=os.getenv('QIANFAN_SK', '')
# )

llm = ChatOpenAI(
    #THUDM/glm-4-9b-chat
    #Qwen/Qwen2.5-7B-Instruct
    model="Qwen/Qwen2.5-7B-Instruct",
    streaming=False,  # 启用流式输出
    api_key=os.getenv('SILICONFLOW_API_KEY', ''), 
    base_url=os.getenv('SILICONFLOW_BASE_URL', ''),
    temperature=0.1,
)
llm_with_tools = llm.bind_tools(tools)

# 创建工具列表的函数版本
functions = [convert_to_openai_function(t) for t in tools]

tools_by_name = {tool.name: tool for tool in tools}

# 定义工具节点函数
def search_tool_node(state: dict):
    result = []
    for tool_call in state["messages"][-1].tool_calls:
        tool = tools_by_name[tool_call["name"]]
        observation = tool.invoke(tool_call["args"])
        print('工具搜索结果的完整输出------>',observation,'\n')
        # 直接转为字符串
        search_result = str(observation)
        result.append(ToolMessage(content=search_result, tool_call_id=tool_call["id"]))
    return {"messages": result}

# 定义流式节点函数
async def chatbot_stream(state: MessagesState):
    """生成流式回复的节点函数"""
    print('chatbot_stream收到的完整输入------>',state,'\n')
    messages = state["messages"]
    # streamed_output = []
    # tool_calls_detected = []  # 新增：保存检测到的工具调用
    
    # 使用非流式方式接收完整返回
    response = llm_with_tools.invoke(
        messages,
        functions=functions,
        function_call="auto"
    )
    return {"messages": [response]}
    
    # 异步生成器不能使用return返回值
    # yield result_state
def route_tools(state: MessagesState) :
    """
    在条件边中使用,如果最后一条消息包含工具调用,则路由到工具节点,否则路由到结束节点。
    """
    messages = state['messages']
    last_message = messages[-1]
    # 检查是否是AI消息且有工具调用
    if hasattr(last_message, 'tool_calls') and last_message.tool_calls:
        # print('last_message.tool_calls------>',last_message.tool_calls,'\n')
        return "search_tool"  # 如果有工具调用，路由到工具节点
    
    return END

def route_chatbot(state: MessagesState) :
    """
    在条件边中使用,如果最后一条消息包含工具调用,则路由到工具节点,否则路由到结束节点。
    """
    messages = state['messages']
    last_message = messages[-1]
    if isinstance(last_message, ToolMessage):
        return "chatbot"
    return END

# 添加chatbot节点
graph_builder.add_node("chatbot", chatbot_stream)
# 添加工具节点
graph_builder.add_node("search_tool", search_tool_node)
# 设置入口点
graph_builder.set_entry_point("chatbot")
graph_builder.add_conditional_edges(
    "chatbot",
    route_tools
)
# 添加从tools到chatbot的边
graph_builder.add_edge("search_tool", "chatbot")
# 编译图
graph = graph_builder.compile()

# 定义一个将图导出为PNG的函数
def export_graph_to_png():
    """
    将LangGraph图导出为PNG格式
    
    Returns:
        str: 生成的PNG文件路径
    """
    try:
        output_file='workflow_graph-' + datetime.now().strftime("%Y-%m-%d_%H-%M-%S") + ".png"
        graph.get_graph().draw_mermaid_png(
            draw_method=MermaidDrawMethod.API,
            output_file_path=output_file
        )
    except Exception as e:
        print(f"导出PNG图形时出错: {e}")
        return None

# 异步运行函数（健壮版本）
async def run_demo():
    """异步运行LangGraph流式输出演示"""
    print("开始流式生成回答...\n")
    
    today = datetime.now().strftime("%Y-%m-%d")
    # print('当前日期:',today,'\n')
    # 创建初始消息
    system_message = SystemMessage(content=f"""
        # 你是一个善于分析的AI助手。
        ## 对于用户的所有问题，先分析是否需要调用搜索工具，再进行回复，比如查询东西，需要调用搜索工具，如果只是问问题，则不需要调用搜索工具。
        ## 请牢记今天的日期是{today},调用工具时，直接使用{today}的日期。
        - 分析示例:比如用户询问日历,星座,黄历,新闻等实时查询相关;比如询问今天的新闻或今天的日历。直接把{today}的日期作为参数传给搜索工具去做搜索。
        - 每个搜索结果都需要按照以下格式返回：
            - 标题：
            - 摘要：
            - 链接：
    """)
    first_message = HumanMessage(content="""
        请帮我搜索下室内适合种植什么花?
    """)
    
    # 初始化状态
    initial_state = {"messages": [system_message, first_message]}
    on_chat_model_stream_list = []
    try:
        # 异步执行流式输出
        async for event in graph.astream_events(initial_state, config={"configurable": {"thread_id": "1"}}, version="v2"):
            # 定义一个变量接收所有on_chat_model_stream的值
            # print('event------>',event,'\n\n')
            event_type = event['event']
            if event_type == 'on_tool_start' and event['data']:
                print('开始调用工具查询', event['data'],'\n\n')
            elif event_type == 'on_tool_end' and event['data']:
                print('工具查询结束',event['data'],'\n\n')
            elif event_type == 'on_chat_model_stream':
                print('on_chat_model_stream事件------>',event["data"]["chunk"].content,'\n\n')
                on_chat_model_stream_list.append(event["data"]["chunk"].content)
    except Exception as e:
        print(f"graph.astream_events执行出错: {e}")
    
    print("\n生成完成！","".join(on_chat_model_stream_list),'\n\n')
    # 展示图形
    try:
        # # 使用官方文档推荐的方法绘制Mermaid图
        # print("使用官方方法绘制Mermaid图...")
        mermaid_diagram = graph.get_graph().draw_mermaid()
        print(f"```mermaid\n{mermaid_diagram}\n```")
        
        # # 导出为PNG
        # print("\n正在导出为PNG图片...")
        export_graph_to_png()
    except Exception as e:
        print(f"图表绘制出错: {e}")
    

# 执行异步函数
if __name__ == "__main__":
    asyncio.run(run_demo())

AI相关

Langgraph

本博客所有文章除特别声明外，均采用 CC BY-SA 4.0 协议，转载请注明出处！

Langgraph学习5：AI爬取工具crawl4ai 上一篇

开发一个本地MCP Server进行code review 下一篇