当前位置：首页 > news >正文

LangGraph(二)——QuickStart样例中的第二步

news 来源：原创 2025/4/23 13:38:47

1. 添加依赖

LangGraph(一)——QuickStart样例中的第一步的基础上添加依赖：
uv add tavily-python langchain-community

2. 官网QuickStart——第二步：用工具增强聊天机器人

2.1 Tavily Search

Tavily提供了实时网页搜索和网页内容提取的功能，官网地址为https://app.tavily.com/home。使用之前需要注册账户，注册后会获取默认的API_KEY。
在这里插入图片描述
计划等级为Researcher，有1000个信用点，收费如下。基础搜索每次查询消耗1信用点，高级搜索每次查询消耗2信用点；基础提取每5个网址消耗1信用点，高级提取每5个网址消耗2信用点。

2.2 简单测试Tavily Search

首先在Windows上设置环境变量TAVILY_API_KEY，可以如下代码测试Tavily Search：

from langchain_community.tools.tavily_search import TavilySearchResultstool = TavilySearchResults(max_results=2)
tools = [tool]
tool.invoke("LangGraph中的node是什么？")

结果：

[{'title': 'LangGraph: Low Level Conceptual Guide 基础概念指南 - 知乎专栏','url': 'https://zhuanlan.zhihu.com/p/716038940','content': 'Nodes 节点. 在LangGraph中，节点通常是Python函数（同步或异步），其中第一个位置参数是state，第二个位置参数（可选）是“config”，包含可选的可配置参数（例如thread_id）。','score': 0.91306627},{'title': '初学LangGraph 之节点、边和状态 - 知乎专栏','url': 'https://zhuanlan.zhihu.com/p/720864252','content': '节点（Nodes）： 是图的基本构建模块 。 每个节点代表一个特定的功能或操作，用来处理当前状态。 节点可以执行计算、修改状态，或者根据接收到的输入生成输出。 简单理解就是在节点内数据的流转是固定的。','score': 0.8706085}]

2.3 添加带工具的ChatBot node

通过bind_tools将指定工具绑定到大语言模型上，目的是为了让大语言模型指导使用工具时要使用何种JSON格式。

rom typing import Annotatedfrom langchain_deepseek import ChatDeepSeek
from typing_extensions import TypedDictfrom langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messagesclass State(TypedDict):messages: Annotated[list, add_messages]graph_builder = StateGraph(State)llm = ChatDeepSeek(model="deepseek-chat")
# Modification: tell the LLM which tools it can call
llm_with_tools = llm.bind_tools(tools)def chatbot(state: State):return {"messages": [llm_with_tools.invoke(state["messages"])]}graph_builder.add_node("chatbot", chatbot)

2.4 添加tool node

创建一个函数以便在调用工具时实际运行tools，并且将该node添加到graph中。下面的代码中实现了BasicToolNode，它检查状态中的最新messages，并在messages包含tool_calls时调用工具。

import jsonfrom langchain_core.messages import ToolMessageclass BasicToolNode:"""A node that runs the tools requested in the last AIMessage."""def __init__(self, tools: list) -> None:self.tools_by_name = {tool.name: tool for tool in tools}def __call__(self, inputs: dict):if messages := inputs.get("messages", []):message = messages[-1]else:raise ValueError("No message found in input")outputs = []for tool_call in message.tool_calls:tool_result = self.tools_by_name[tool_call["name"]].invoke(tool_call["args"])outputs.append(ToolMessage(content=json.dumps(tool_result, ensure_ascii=False),name=tool_call["name"],tool_call_id=tool_call["id"],))return {"messages": outputs}tool_node = BasicToolNode(tools=[tool])
graph_builder.add_node("tools", tool_node)

2.5 添加条件边

edge将控制流从一个node导向下一个node。conditional edge包含if语句，根据当前graph的state导向不同的node。定义conditional edge的函数接收当前graph的state，并返回一个字符串或字符串列表，指示下一步调用哪个node。
下面的代码中定义了一个名为route_tools的路由函数，该函数检查ChatBot输出中的tool_calls，如果tool_calls存在，该函数将graph从当前node导向到tools，否则将graph从当前node导向到END。通过调用add_conditional_edges对graph添加从chatbot到tools和从chatbot到END的两条condtional edges。

def route_tools(state: State,
):"""Use in the conditional_edge to route to the ToolNode if the last messagehas tool calls. Otherwise, route to the end."""if isinstance(state, list):ai_message = state[-1]elif messages := state.get("messages", []):ai_message = messages[-1]else:raise ValueError(f"No messages found in input state to tool_edge: {state}")if hasattr(ai_message, "tool_calls") and len(ai_message.tool_calls) > 0:return "tools"return END# The `tools_condition` function returns "tools" if the chatbot asks to use a tool, and "END" if
# it is fine directly responding. This conditional routing defines the main agent loop.
graph_builder.add_conditional_edges("chatbot",route_tools,# The following dictionary lets you tell the graph to interpret the condition's outputs as a specific node# It defaults to the identity function, but if you# want to use a node named something else apart from "tools",# You can update the value of the dictionary to something else# e.g., "tools": "my_tools"{"tools": "tools", END: END},
)
# Any time a tool is called, we return to the chatbot to decide the next step
graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge(START, "chatbot")
graph = graph_builder.compile()

2.6 可视化StateGraph

通过如下的代码可视化graph：

from IPython.display import Image, displaytry:display(Image(graph.get_graph().draw_mermaid_png()))
except Exception:# This requires some extra dependencies and is optionalpass

可视化结果：
在这里插入图片描述

2.7 构建聊天循环

通过如下的代码构建聊天循环：

def stream_graph_updates(user_input: str):for event in graph.stream({"messages": [{"role": "user", "content": user_input}]}):for value in event.values():print("Assistant:", value["messages"][-1].content)while True:try:user_input = input("User: ")if user_input.lower() in ["quit", "exit", "q"]:print("Goodbye!")breakstream_graph_updates(user_input)except:# fallback if input() is not availableuser_input = "你知道什么关于LangGraph的内容？"print("User: " + user_input)stream_graph_updates(user_input)break

输入“搜索关于LLM-Based Agent的方法论或实现技巧并总结”后，会调用Tavily Search，得到如下两条结果(前文定义工具的时候限定了返回最大结果数为2)：
在这里插入图片描述
在获取完Tavily Search的结果，会重新返回大模型，进行总结工作：

Assistant: 以下是关于LLM-Based Agent的方法论和实现技巧的总结：### 方法论
1. **任务分解**：- 通过简单的提示（如“XYZ的步骤”或“实现XYZ的子目标是什么”）由LLM完成。- 使用特定任务的指令，例如“为写文章生成大纲”。2. **对话管理**：- **上下文追踪**：需要正确记录用户的历史意图、槽位值和完整的对话历史，确保Agent具备深层次的理解和回应能力。- **状态更新**：动态更新对话状态，保持信息一致性。可以采用高效的算法和数据结构优化更新过程。- **意图更新**：结合历史意图和最新对话轮次的内容，综合判断用户的当前意图。3. **槽位追踪**：- 通过设计的提示语引导用户补充遗漏的槽位信息，提高任务完成率。- 例如，用户未提供送餐地址时，Agent可以生成提示：“请问需要送到什么地址呢？”4. **智能调用功能**：- 将用户的意图和槽位值转换为具体操作，完成用户需求。- 需要关注高效和准确的设计范式。### 实现技巧
1. **对话历史剪裁**：- 对话历史的长度不宜过长（例如不超过3轮），以避免模型产生幻觉。- 在结合参考资料生成内容时（如RAG），需注意内容的正确性，可能需要引入前置分类器或验证机制。2. **状态存储**：- 小规模应用可将状态存储在内存中。- 大规模分布式应用需使用持久化存储（如Redis），通过session ID快速读写对话状态。3. **测试与验证**：- 在面向客户的场景中，需进行大量测试（单元测试、集成测试、用户接受测试）以确保模型的稳定性和准确性。- 持续监控和优化模型，及时发现并解决问题。4. **提示设计**：- 使用自然、不显生硬的提示语（如“请问...”或“我们需要...”），提升用户体验。通过这些方法论和技巧，可以更高效地设计和实现基于LLM的Agent，提升其在任务型对话中的表现。
Goodbye!