langgraph agent chat ui搭建与使用教程

在构建复杂的 AI Agent 时，仅靠命令行日志难以直观评估多步推理的效果。搭建一个专用的 Chat UI 不仅能实时展示 Agent 的思考过程（如工具调用、中间状态），还能让非技术人员直接参与测试。本教程将演示如何结合 LangGraph 的状态机能力与 Streamlit 的轻量级前端，快速构建一个支持流式响应和状态回溯的对话界面。

核心原理在于将 LangGraph 的 StateGraph 作为后端引擎，通过 API 或直接函数调用与前端组件通信。前端负责收集用户输入并渲染历史消息，后端负责维护对话上下文并执行节点逻辑。这种分离架构确保了业务逻辑与界面展示的解耦，便于后续扩展。

开始前，请确保已安装 Python 3.9+，并准备好以下库：langgraph, langchain-core, streamlit。建议使用虚拟环境隔离依赖。

LangGraph 后端与 Streamlit 前端的数据交互流程

1. 定义 LangGraph 状态与节点

首先需要一个基础的 LangGraph 应用作为后端。我们定义一个简单的 ReAct 风格 Agent，包含“思考”和“回答”两个节点，并使用 MessagesState 管理对话历史。

创建 agent_graph.py 文件，引入必要的模块并定义状态结构。这里使用 TypedDict 明确状态中包含的消息列表。

from typing import Annotated, Sequence, TypedDict
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langgraph.graph import StateGraph, END
import operator

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]

def think_node(state: AgentState):
    # 模拟思考过程，实际项目中可替换为 LLM 调用
    last_message = state['messages'][-1]
    return {"messages": [AIMessage(content=f"正在处理: {last_message.content}")]}

def respond_node(state: AgentState):
    # 模拟最终回答
    return {"messages": [AIMessage(content="这是基于 LangGraph 生成的回答。")]}

def build_graph():
    workflow = StateGraph(AgentState)
    workflow.add_node("think", think_node)
    workflow.add_node("respond", respond_node)
    workflow.set_entry_point("think")
    workflow.add_edge("think", "respond")
    workflow.add_edge("respond", END)
    return workflow.compile()

这段代码构建了一个线性工作流：用户输入 -> 思考节点 -> 回答节点 -> 结束。关键在于 operator.add 的使用，它确保新消息追加到现有列表中，而不是覆盖，从而保留完整的对话上下文。

1. 定义 LangGraph 状态与节点

2. 搭建 Streamlit 前端骨架

接下来创建 app.py，使用 Streamlit 构建聊天界面。Streamlit 的 st.chat_message 和 st.chat_input 组件能极大简化 UI 开发。

初始化页面配置并设置侧边栏用于显示调试信息，这对观察 Agent 内部状态非常有用。

import streamlit as st
from agent_graph import build_graph

st.set_page_config(page_title="LangGraph Agent Chat", layout="wide")
st.title("🤖 LangGraph Agent 交互演示")

# 初始化 Session State
if "messages" not in st.session_state:
    st.session_state.messages = []
if "graph" not in st.session_state:
    st.session_state.graph = build_graph()

此时运行 streamlit run app.py，你将看到一个空的聊天窗口。下一步是将前端的输入与后端的图执行连接起来。

Streamlit 应用启动后的初始空白聊天界面

3. 连接输入与图执行逻辑

在 Streamlit 主循环中，监听用户的聊天输入。当用户发送消息时，将其封装为 HumanMessage 并传入 LangGraph。

注意，我们需要将 st.session_state.messages 中的历史消息转换为 LangChain 兼容的格式，以便 Graph 能正确读取上下文。

# 显示历史消息
for message in st.session_state.messages:
    with st.chat_message(message["role"]):
        st.markdown(message["content"])

# 处理用户输入
if prompt := st.chat_input("请输入你的问题..."):
    # 1. 显示用户消息
    st.session_state.messages.append({"role": "user", "content": prompt})
    with st.chat_message("user"):
        st.markdown(prompt)

    # 2. 准备 Graph 输入状态
    from langchain_core.messages import HumanMessage
    input_state = {
        "messages": [HumanMessage(content=prompt)]
    }
    
    # 3. 执行 Graph
    with st.chat_message("assistant"):
        placeholder = st.empty()
        full_response = ""
        
        # 这里为了演示简单，直接 invoke；生产环境建议使用 stream 模式
        config = {"configurable": {"thread_id": "test-1"}}
        result = st.session_state.graph.invoke(input_state, config=config)
        
        # 获取最后一条 AI 消息
        ai_message = result["messages"][-1].content
        placeholder.markdown(ai_message)
        
    # 4. 保存助手回复到历史
    st.session_state.messages.append({"role": "assistant", "content": ai_message})

这一步实现了基本的请求-响应循环。config 参数中的 thread_id 对于 LangGraph 至关重要，它决定了状态存储在哪个会话线程中，从而实现多用户隔离或长对话记忆。

4. 实现流式输出与中间状态可见性

为了让体验更流畅，我们修改执行部分以支持流式输出，并在侧边栏实时展示 Agent 的中间步骤（如工具调用详情）。

使用 graph.stream 方法代替 invoke，它可以逐个产出节点执行结果。

    # 替换之前的 invoke 部分
    with st.chat_message("assistant"):
        message_placeholder = st.empty()
        full_response = ""
        
        # 用于在侧边栏显示中间状态
        debug_container = st.sidebar.container()
        debug_container.write("**执行日志:**")
        
        for event in st.session_state.graph.stream(input_state, config=config):
            for node_name, node_output in event.items():
                # 更新侧边栏调试信息
                with debug_container:
                    st.json({"node": node_name, "output_preview": str(node_output)[:100]})
                
                # 如果是最终回答节点，累积内容
                if node_name == "respond":
                    content = node_output["messages"][-1].content
                    full_response += content
                    message_placeholder.markdown(full_response + "▌")
        
        message_placeholder.markdown(full_response)

聊天过程中侧边栏实时显示节点执行日志

通过这种方式，用户不仅能看到最终答案，还能在侧边栏观察到 Agent 经过了哪些节点，每个节点输出了什么数据。这对于调试复杂的条件分支逻辑非常有帮助。

5. 验证与状态持久化测试

完成编码后，启动应用进行全流程测试。输入一个复杂问题，观察侧边栏是否按顺序显示了 think 和 respond 节点的日志。

尝试刷新页面，由于我们使用了 st.session_state 存储前端历史，但 LangGraph 的后端状态通常依赖于检查点（Checkpointer）。若要实现真正的断点续聊，需在 build_graph 中配置 MemorySaver 或 SqliteSaver。

from langgraph.checkpoint.memory import MemorySaver

def build_graph_with_memory():
    memory = MemorySaver()
    workflow = StateGraph(AgentState)
    # ... 添加节点和边 ...
    return workflow.compile(checkpointer=memory)

启用 Checkpointer 后，即使重启后端服务，只要 thread_id 不变，Agent 仍能记住之前的对话内容。在前端，你可以增加一个“清除历史”按钮，通过清空 st.session_state.messages 并生成新的 thread_id 来重置会话。

多轮对话后界面展示的历史记录与当前状态

最终，你获得了一个具备完整交互能力的 Agent 界面：左侧是流畅的聊天窗口，右侧是透明的执行黑盒。这种结构既满足了最终用户的易用性需求，也保留了开发者所需的调试深度。工作文件应保留 agent_graph.py 和 app.py，以便后续接入真实的 LLM 模型和工具集。