LangGraph--基础学习（Human-in-the-loop 人工参与循环）
简单来说，智能体无法区分内容的好坏，我们设计智能体的人需要考虑到需要用户接收哪些信息，哪些不需要用户接收，或者让用户做判断的，如果我们设计者可以提取判断，这可以通过这个环节进行解决，怎么解决呢？很简单，当智能体打算调用大模型可以进行检测输入是否符合预期，工具结果返回是否达到我们设计者的预期，如果无法达到，则返回我们人为定义的数据，这就是所谓的人工参与，另一层含义就是智能体的最终决策由人类控制，但是内部实现都是一样的，下面就详细介绍：
LangGraph 支持强大的人在环（HIL） 工作流，在自动化流程的任何点都可以进行人工干预。这在大型语言模型（LLM）驱动的应用程序中特别有用，因为在这些应用程序中，模型输出可能需要验证、校正或其他上下文。
持久执行状态 ：LangGraph 在每个步骤后检查图状态，允许执行在定义的节点处无限期暂停。这支持异步人工审查或输入，而没有时间限制。
灵活的集成点 ：HIL 逻辑可以在工作流程中的任何点引入。这允许有针对性的人工参与，例如批准 API 调用、纠正输出或指导对话。
典型用例 ¶
🛠️审查工具调用 ：人类可以在工具执行之前审查，编辑或批准 LLM 请求的工具调用。
验证 LLM 输出 ：人类可以审查、编辑或批准 LLM 生成的内容。
提供上下文 ：使 LLM 能够明确请求人类输入以澄清或其他细节，或支持多轮对话。
实现
interrupt ：在特定点暂停执行，提供信息供人查看。
Command：用于使用人类提供的值恢复执行。
例子：
import os
from dotenv import load_dotenv# 加载.env文件中的环境变量
load_dotenv()
from typing import Annotated,List,Literal
from typing_extensions import TypedDict
from langchain_deepseek import ChatDeepSeek
from langchain_tavily import TavilySearch
from langchain_core.messages import BaseMessage,SystemMessage,AIMessage,HumanMessage,ToolMessage
from typing_extensions import TypedDict
from langgraph.graph import StateGraph,START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
from langchain_core.tools import tool
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import Command, interrupt
from pydantic import BaseModel
from langgraph.checkpoint.memory import MemorySaver
memory = MemorySaver()
from langchain_tavily import TavilySearch
from langchain_community.tools.tavily_search import TavilySearchResults
from langgraph.prebuilt import ToolNode,tools_condition class State(TypedDict):messages:Annotated[list, add_messages] # 增加不覆盖llm = ChatDeepSeek(model="deepseek-chat",api_key=os.getenv("DEEPSEEK_API_KEY"))
tool = TavilySearchResults(max_results=2)
tool_node =ToolNode(tools=[tool])
tools =[tool]
llm_with_tools = llm.bind_tools(tools)from langchain_core.messages import BaseMessage,SystemMessage,AIMessage,HumanMessage,ToolMessage
def chatbot(state:State):message_content = state["messages"][0].contentmessage = [SystemMessage(content="你是一个聊天机器人"),HumanMessage(content=message_content)]return {"messages":[llm_with_tools.invoke(message)]}graph_builder = StateGraph(State)
graph_builder.add_node("chatbot",chatbot)
graph_builder.add_node("tools", tool_node)
graph_builder.add_edge(START,"chatbot")
graph_builder.add_conditional_edges("chatbot",tools_condition,["tools"])
graph_builder.add_edge("tools","chatbot")
graph_builder.add_edge("chatbot",END)graph = graph_builder.compile(checkpointer=memory,interrupt_before=["tools"]
)
# 打印图结构
print(graph.get_graph().draw_mermaid())
graph_png = graph.get_graph().draw_mermaid_png()
with open("human_in_loop.png", "wb") as f:f.write(graph_png)
config = {"configurable":{"thread_id":"4"}}
user_input = "我想学习langgraph，你有什么好的建议吗？"
events = graph.stream({"messages":[("user",user_input)]},config)
for event in events:if"messages" in event:event["messages"][-1].pretty_print()snapshot = graph.get_state(config)
snapshot.next
('tools',)existing_message = snapshot.values["messages"][-1]
existing_message.pretty_print()================================== Ai Message ==================================
Tool Calls:tavily_search_results_json (call_0_ef66d138-4949-4e47-a1ac-f49552d6e0f0)Call ID: call_0_ef66d138-4949-4e47-a1ac-f49552d6e0f0Args:query: 学习langgraph的建议from langchain_core.messages import AIMessage,ToolMessage
# 使用下面的回答直接代替工具返回的数据
answer = ("langgraph是比较安全的lib，现在应用广泛应用到agent上了")
new_message = [ToolMessage(content=answer,tool_call_id= existing_message.tool_calls[0]["id"]),AIMessage(content=answer),]
new_message[-1].pretty_print()================================== Ai Message ==================================langgraph是比较安全的lib，现在应用广泛应用到agent上了graph.update_state(config,{"messages":new_message},as_node="tools"){'configurable': {'thread_id': '4','checkpoint_ns': '','checkpoint_id': '1f04f511-cc15-6946-8004-f67f17bd4c97'}}
snapshot = graph.get_state(config)
after_update_messages = snapshot.values["messages"][-1]
after_update_messagesAIMessage(content='langgraph是比较安全的lib，现在应用广泛应用到agent上了', additional_kwargs={}, response_metadata={}, id='39d280d9-92f3-4d86-8746-f056d95dc9a8')
snapshot.next
('chatbot',)
events = graph.stream(None, config,stream_mode="values")
for event in events:if "messages" in event:event["messages"][-1].pretty_print()