Agent开发实战-langgraph重构

observer？

379人浏览 · 2026-04-22 14:37:20

observer？ · 2026-04-22 14:37:20 发布

LangGraph 版紫微斗数 AI Agent 深度解析

引言

在 AI 大模型时代，构建一个智能、可靠的对话系统已经成为许多开发者的目标。而 LangGraph 作为一个新兴的工作流编排框架，为我们提供了一种全新的方式来构建复杂的 AI 应用。本文将详细解析基于 LangGraph 实现的紫微斗数 AI Agent，探讨其技术架构、核心功能和实现细节。

LangGraph 简介

LangGraph 是一个基于有向图（Directed Graph）的工作流编排框架，它允许开发者将复杂的 AI 应用分解为多个节点（Nodes），并通过边（Edges）定义它们之间的流转关系。这种基于图的方法使得 AI 应用的逻辑更加清晰、可维护，并且能够更好地处理复杂的多步骤任务。

核心概念

State：全局状态，存储整个工作流的上下文信息
Nodes：节点，执行具体的任务（如调用大模型、执行工具等）
Edges：边，定义节点之间的流转关系
Graph：图，由节点和边组成的完整工作流

langgraph_agent.py 架构分析

1. 整体架构

langgraph_agent.py 采用了模块化的设计，主要包含以下几个部分：

工具定义：使用 @tool 装饰器定义工具
记忆系统：基于 Chroma 向量数据库实现的长期记忆
大模型集成：使用本地 Qwen 模型
状态定义：使用 TypedDict 定义全局状态
节点定义：包括记忆检索、大脑思考和记忆存储节点
图网络构建：使用 StateGraph 构建工作流
主循环：处理用户输入并运行图网络

2. 核心组件

2.1 记忆系统

print("[*] 正在初始化长期记忆系统...")
EMBEDDING_MODEL_PATH = "./local_models/bge-small-zh-v1.5"
LONG_TERM_DB_DIR = "./long_term_memory_db"

try:
    device = get_device()
    embeddings = HuggingFaceEmbeddings(
        model_name=EMBEDDING_MODEL_PATH, 
        model_kwargs={'device': device}
    )
    long_term_db = Chroma(
        collection_name="user_memory", 
        persist_directory=LONG_TERM_DB_DIR, 
        embedding_function=embeddings
    )
    print("[OK] 长期记忆系统初始化成功")
except Exception as e:
    print(f"[ERROR] 长期记忆系统初始化失败: {str(e)}")
    long_term_db = None

技术亮点：

使用 Chroma 向量数据库存储长期记忆
支持设备自动选择（CPU/GPU）
完善的错误处理机制

2.2 状态定义

class AgentState(TypedDict):
    # Annotated 与 add_messages 会自动把新消息追加到列表中，不用再手动 append 了！
    messages: Annotated[list, add_messages]
    # 记忆相关字段
    memory_context: str = ""

技术亮点：

使用 TypedDict 定义类型安全的状态结构
使用 Annotated[list, add_messages] 实现自动消息追加
添加 memory_context 字段存储记忆检索结果

2.3 节点定义

def memory_retrieval_node(state: AgentState):
    """记忆检索节点：检索与当前问题相关的历史记忆"""
    print("\n[🧠 节点执行: 记忆检索] 正在检索相关记忆...")
    memory_context = ""
    
    # 提取用户最新问题
    latest_user_msg = None
    for msg in reversed(state["messages"]):
        if hasattr(msg, 'type') and msg.type == "human":
            latest_user_msg = msg
            break
    
    # 检索相关记忆
    if latest_user_msg and long_term_db:
        try:
            related_memories = long_term_db.similarity_search(latest_user_msg.content, k=1)
            if related_memories:
                memory_context = f"\n\n[相关记忆]: {related_memories[0].page_content}"
                print("[OK] 检索到相关记忆")
            else:
                print("[INFO] 未检索到相关记忆")
        except Exception as e:
            print(f"[ERROR] 记忆检索失败: {str(e)}")
    
    return {"memory_context": memory_context}

技术亮点：

基于语义相似度检索相关记忆
完善的错误处理
清晰的日志输出

2.4 图网络构建

print("[*] 正在编排 LangGraph 工作流图...")
graph_builder = StateGraph(AgentState)

# 添加节点
graph_builder.add_node("memory_retrieval", memory_retrieval_node)
graph_builder.add_node("oracle", chatbot_node)
graph_builder.add_node("memory_storage", memory_storage_node)

# 定义边：起点 -> 记忆检索 -> 大脑 -> 记忆存储 -> 结束
graph_builder.add_edge(START, "memory_retrieval")
graph_builder.add_edge("memory_retrieval", "oracle")
graph_builder.add_edge("oracle", "memory_storage")
graph_builder.add_edge("memory_storage", END)

# 编译图网络！(此时它变成了一个可执行的 Agent)
app = graph_builder.compile()

技术亮点：

使用 StateGraph 构建有向图
清晰的节点和边定义
模块化的工作流设计

工作流程分析

用户输入：用户输入问题
记忆检索：系统检索与当前问题相关的历史记忆
大脑思考：大模型基于问题和记忆生成回答
记忆存储：系统将对话保存到长期记忆
输出回答：系统向用户展示最终回答

工作流程图

[用户输入] → [记忆检索节点] → [大脑节点] → [记忆存储节点] → [输出回答]

技术优势

1. 模块化设计

节点化：每个功能都封装在独立的节点中，便于维护和扩展
边定义：清晰的边定义使得工作流逻辑一目了然
状态管理：统一的状态管理机制，便于调试和监控

2. 记忆增强

短期记忆：使用 LangGraph 的 add_messages 自动管理
长期记忆：使用 Chroma 向量数据库持久化存储
记忆检索：基于语义相似度检索相关历史记忆
记忆应用：将检索到的记忆整合到系统提示中

3. 鲁棒性

错误处理：完善的异常处理机制
设备适配：自动选择 CPU 或 GPU
日志输出：清晰的日志输出，便于调试

4. 性能优化

流式执行：使用 stream 方法实现流式执行，提高响应速度
内存管理：合理的内存使用，避免内存泄漏
计算优化：根据硬件情况选择最佳计算设备

使用示例

基本使用

启动服务：确保本地 Qwen 模型服务正在运行
运行脚本：
```
python langgraph_agent.py
```

对话示例：

🧑‍💻 你: 我是1997年3月13日出生的，分析我的命盘
[🧠 节点执行: 记忆检索] 正在检索相关记忆...
[INFO] 未检索到相关记忆

[🧠 节点执行: 大脑] 正在思考...

✨ 大师最终回答: 根据您的出生信息，我为您分析如下命盘...

[🧠 节点执行: 记忆存储] 正在保存对话到长期记忆...
[OK] 对话已保存到长期记忆

🧑‍💻 你: 我是什么命格
[🧠 节点执行: 记忆检索] 正在检索相关记忆...
[OK] 检索到相关记忆

[🧠 节点执行: 大脑] 正在思考...

✨ 大师最终回答: 根据之前的分析，您的命格为...

[🧠 节点执行: 记忆存储] 正在保存对话到长期记忆...
[OK] 对话已保存到长期记忆