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 采用了模块化的设计，主要包含以下几个部分：

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

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. 记忆检索：系统检索与当前问题相关的历史记忆
  3. 大脑思考：大模型基于问题和记忆生成回答
  4. 记忆存储：系统将对话保存到长期记忆
  5. 输出回答：系统向用户展示最终回答

工作流程图

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

技术优势

1. 模块化设计

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

2. 记忆增强

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

3. 鲁棒性

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

4. 性能优化

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

使用示例

基本使用

  1. 启动服务：确保本地 Qwen 模型服务正在运行
  2. 运行脚本： 
    python langgraph_agent.py
  3. 对话示例： 
    🧑‍💻 你: 我是1997年3月13日出生的，分析我的命盘
[🧠 节点执行: 记忆检索] 正在检索相关记忆...
[INFO] 未检索到相关记忆

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

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

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

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

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

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

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

高级配置

  • 调整记忆参数：修改 LONG_TERM_DB_DIR 和 EMBEDDING_MODEL_PATH
  • 调整模型参数：修改 ChatOpenAI 的配置
  • 扩展节点：添加新的节点和边，实现更复杂的功能

代码优化建议

1. 增强记忆管理

  • 记忆评分：为记忆添加评分机制，优先保留更重要的记忆
  • 记忆压缩：实现记忆压缩算法，减少存储开销
  • 记忆过期：添加记忆过期机制，自动清理过时的记忆

2. 增强工具调用

  • 多工具支持：添加更多工具，如天气查询、日历管理等
  • 工具选择优化：实现智能工具选择机制，提高工具使用效率
  • 工具结果缓存：缓存工具调用结果，避免重复调用

3. 增强错误处理

  • 智能错误分类：根据错误类型提供不同的处理策略
  • 错误预防：在系统提示中添加错误预防指南
  • 错误恢复：实现更强大的错误恢复机制

4. 增强用户体验

  • 多模态支持：添加图片、语音等多模态输入输出
  • 个性化设置：允许用户自定义 Agent 的性格和风格
  • 实时反馈：提供更实时的执行状态反馈

应用前景

LangGraph 版紫微斗数 AI Agent 展示了如何使用现代 AI 框架构建复杂的工作流。这种基于图的方法为构建更智能、更可靠的 AI 系统提供了强大的工具。

潜在应用场景

  1. 智能客服：构建具有记忆能力的智能客服系统
  2. 个人助手：开发能够记住用户偏好的个人助手
  3. 教育辅导：创建能够适应学生学习进度的教育辅导系统
  4. 医疗咨询：开发能够记住患者历史的医疗咨询系统
  5. 金融顾问：构建能够分析用户财务历史的金融顾问系统

结论

LangGraph 版紫微斗数 AI Agent 是一个结合了现代 AI 技术和传统命理文化的创新应用。它不仅展示了 LangGraph 框架的强大能力，也为我们提供了一种构建智能、可靠、具有记忆能力的 AI 系统的方法。

通过模块化的设计、强大的记忆系统和灵活的工作流编排，我们可以构建出更加智能、更加人性化的 AI 应用。随着技术的不断发展，相信这种基于图的 AI 工作流方法将会在更多领域得到应用，为我们的生活和工作带来更多便利。

# 重构 # nlp # python # 机器学习
Logo
DeepSeek技术社区

欢迎加入DeepSeek 技术社区。在这里，你可以找到志同道合的朋友，共同探索AI技术的奥秘。

更多推荐

GPT-6 来了，这次可能会重新定义 ChatGPT 的用法

帮我写一篇 CSDN 风格文章，标题是《GPT-6 来了，这次可能会重新定义 ChatGPT 的用法》，读者是程序员和 AI 用户，前半段讲趋势，中间讲使用场景，后半段自然带出 Plus、Pro、Codex，不要太硬广，语气像经验分享。你说“写给程序员看”，它要知道不能写太小白，要讲真实开发场景，比如调试、重构、测试、接口、项目维护。真实开发里面，有需求分析、项目理解、代码结构、接口设计、异常处理

avatar DeepSeek技术社区
cover

主流大模型矩阵对比（Claude/Codex/Gemini 等）

avatar DeepSeek技术社区
cover

31.MoE 架构深度解析：DeepSeek、Mixtral 背后的稀疏化魔法

avatar DeepSeek技术社区