图片生成 LangGraph 迁移计划

一、背景与目标

当前架构

聊天流程：api/chat.py → langgraph_chat.py → ChatOpenAI → OpenRouter
图片流程：api/image.py → model_integration.py → OpenAI Client → OpenRouter

问题：

图片生成流程独立，无法与聊天共享上下文
缺乏状态管理，无法支持多轮图片编辑
架构不统一，后续扩展 Agent 能力困难

目标架构

聊天流程：api/chat.py → LangGraphChatService → ChatOpenAI → OpenRouter
图片流程：api/image.py → LangGraphImageService → model_integration → OpenRouter
                                    ↓
                          共享 MySQL Checkpointer
                          同一 thread_id 共享对话历史

收益：

✅ 对话式图片编辑：用户可在对话中说"生成一张图"，然后说"把背景改成蓝色"
✅ 多步骤图片处理：支持生成 → 编辑 → 优化的链式流程
✅ 统一架构：便于后续添加 Agent 工具能力

二、实施阶段

Phase 1：基础架构迁移（优先）

目标：将图片生成逻辑迁移到 LangGraph，共享 checkpoint

1.1 新增服务类

在 services/langgraph_chat.py 中新增 LangGraphImageService 类：

python

class LangGraphImageService:
    """LangGraph 图片生成服务

    使用 LangGraph 管理图片生成流程，支持：
    - 共享 checkpoint（与聊天使用同一 thread_id）
    - 流式输出（可选）
    - 后续扩展工具调用能力
    """

    def __init__(self, config: Optional[ChatConfig] = None):
        self.config = config or ChatConfig()
        self.model_integration = get_model_integration()

    def generate_image(
        self,
        thread_id: str,
        prompt: str,
        model: str,
        reference_images: Optional[List[str]] = None,
        user_info: Optional[Dict[str, Any]] = None
    ) -> Dict[str, Any]:
        """生成图片（带 checkpoint 管理）"""
        pass

    def generate_image_stream(
        self,
        thread_id: str,
        prompt: str,
        model: str,
        reference_images: Optional[List[str]] = None,
        user_info: Optional[Dict[str, Any]] = None
    ) -> Iterator[Dict[str, Any]]:
        """流式生成图片（可选，用于进度反馈）"""
        pass

    def _build_image_graph(self, checkpointer, model: str):
        """构建图片生成 Graph"""
        # 节点：处理参考图 → 调用生图 API → 返回结果
        pass

1.2 Graph 结构设计

python

# 定义状态
class ImageState(TypedDict):
    prompt: str
    model: str
    reference_images: Optional[List[str]]
    generated_images: Optional[List[str]]  # base64 URLs
    error: Optional[str]

# 构建 Graph
workflow = StateGraph(ImageState)
workflow.add_node("prepare", prepare_node)
workflow.add_node("generate", generate_node)
workflow.add_node("save", save_node)
workflow.add_edge(START, "prepare")
workflow.add_edge("prepare", "generate")
workflow.add_edge("generate", "save")
workflow.add_edge("save", END)

1.3 修改 API 层

修改 api/image.py，调用新的 LangGraph 服务：

python

# 修改前
from core.extensions import model_integration
base64_urls = model_integration.generate_image(...)

# 修改后
from services.langgraph_chat import get_image_service
result = get_image_service().generate_image(
    thread_id=thread_id,
    prompt=prompt,
    model=model,
    reference_images=llm_ref_urls,
    user_info=user_info
)
base64_urls = result["generated_images"]

1.4 保留 model_integration 作为底层调用

model_integration.py 保留，作为被 LangGraph 节点调用的工具：

LangGraphImageService
    → generate_node
        → model_integration.generate_image()  # 底层 API 调用

Phase 2：多步骤图片处理

目标：添加图片编辑工具，支持链式处理

2.1 定义图片处理工具

python

# services/image_tools.py
from langchain_core.tools import tool

@tool
def generate_image(prompt: str, model: str) -> str:
    """生成图片"""
    pass

@tool
def edit_image(image_url: str, instruction: str) -> str:
    """编辑图片（如修改背景、添加元素）"""
    pass

@tool
def apply_filter(image_url: str, filter_type: str) -> str:
    """应用滤镜"""
    pass

@tool
def upscale_image(image_url: str, scale: int = 2) -> str:
    """放大图片"""
    pass

2.2 扩展 Graph 结构

Phase 3：对话式图片编辑

目标：统一聊天和图片生成，支持意图识别和自动路由

3.1 意图识别节点

python

@tool
def detect_intent(user_message: str) -> str:
    """识别用户意图：chat / generate_image / edit_image"""
    pass

3.2 统一 Graph（可选）

三、涉及文件

文件	操作	说明
`services/langgraph_chat.py`	修改	新增 `LangGraphImageService` 类
`services/image_tools.py`	新增	图片处理工具定义（Phase 2）
`api/image.py`	修改	改为调用 LangGraph 服务
`services/model_integration.py`	保留	作为底层 API 调用工具

四、技术要点

4.1 共享 Checkpoint

图片生成和聊天使用同一 thread_id，共享对话历史：

python

# 聊天时
chat_service.chat_stream(thread_id="conv_123", ...)

# 图片生成时（同一个 thread_id）
image_service.generate_image(thread_id="conv_123", ...)

Checkpoint 中会保存完整的消息历史，包括图片生成的结果。

4.2 状态设计

python

class ImageState(TypedDict):
    messages: List[BaseMessage]  # 继承自 MessagesState，保持兼容
    prompt: str
    model: str
    reference_images: Optional[List[str]]
    generated_images: Optional[List[str]]
    current_step: str  # generate / edit / filter / done
    error: Optional[str]

4.3 流式输出（可选）

图片生成通常需要较长时间，可通过 SSE 返回进度：

python

def generate_image_stream(self, ...):
    yield {"status": "processing", "message": "正在生成图片..."}
    # 调用 API
    yield {"status": "done", "images": [...]}

五、时间规划

阶段	预计工作量	依赖
Phase 1	基础架构迁移	1-2 天
Phase 2	多步骤处理	2-3 天
Phase 3	对话式编辑	3-5 天

语言特性比对

_posts

CloudNative

sakila

图片生成 LangGraph 迁移计划

一、背景与目标

当前架构

目标架构

二、实施阶段

Phase 1：基础架构迁移（优先）

1.1 新增服务类

1.2 Graph 结构设计

1.3 修改 API 层

1.4 保留 model_integration 作为底层调用

Phase 2：多步骤图片处理

2.1 定义图片处理工具

2.2 扩展 Graph 结构

Phase 3：对话式图片编辑

3.1 意图识别节点

3.2 统一 Graph（可选）

三、涉及文件

四、技术要点

4.1 共享 Checkpoint

4.2 状态设计

4.3 流式输出（可选）

五、时间规划

六、参考资料

图片生成 LangGraph 迁移计划 ​

一、背景与目标 ​

当前架构 ​

目标架构 ​

二、实施阶段 ​

Phase 1：基础架构迁移（优先） ​

1.1 新增服务类 ​

1.2 Graph 结构设计 ​

1.3 修改 API 层 ​

1.4 保留 model_integration 作为底层调用 ​

Phase 2：多步骤图片处理 ​

2.1 定义图片处理工具 ​

2.2 扩展 Graph 结构 ​

Phase 3：对话式图片编辑 ​

3.1 意图识别节点 ​

3.2 统一 Graph（可选） ​

三、涉及文件 ​

四、技术要点 ​

4.1 共享 Checkpoint ​

4.2 状态设计 ​

4.3 流式输出（可选） ​

五、时间规划 ​

六、参考资料 ​

图片生成 LangGraph 迁移计划

一、背景与目标

当前架构

目标架构

二、实施阶段

Phase 1：基础架构迁移（优先）

1.1 新增服务类

1.2 Graph 结构设计

1.3 修改 API 层

1.4 保留 model_integration 作为底层调用

Phase 2：多步骤图片处理

2.1 定义图片处理工具

2.2 扩展 Graph 结构

Phase 3：对话式图片编辑

3.1 意图识别节点

3.2 统一 Graph（可选）

三、涉及文件

四、技术要点

4.1 共享 Checkpoint

4.2 状态设计

4.3 流式输出（可选）

五、时间规划

六、参考资料