LangGraph —— 實戰智能編輯 Agent

  1. 需求描述 =======

在前面課程的《LangGraph 快速入門》中,我們已經介紹了 LangGraph 框架的核心概念和基礎用法,相信大家一定已經躍躍欲試,想要立即動手來實現一個 Agent 了。那麼本節課,我們就直接上乾貨,實現一個同時具備寫文章 + 配插圖能力的智能編輯 Agent。

在寫代碼之前,我們先來簡單梳理下業務需求。

回想一下我們平時寫文章或者技術博客的大概流程:首先要確定一個主題,然後想一個博人眼球的標題,接下來就是寫正文,最後再配上一張生動有趣的插圖。整個流程如下圖所示:

看到這個流程圖,詳細細心的同學一定發現了:這不就是個標準的工作流 Workflow 嘛!現在大模型的創作能力已經這麼強了,我是不是可以基於大模型搭建一個 Agent,一鍵完成上面的全部流程呢?當然可以,這就是 LangGraph 最擅長做的事情。

那麼閒話少敘,我們馬上就用 LangGraph 來實現這樣一個智能編輯 Agent。多代碼預警!

  1. 狀態定義 =======

用 LangGraph 實現 Agent,最核心的環節就是定義 State 狀態,它是多個節點間協作和通信的關鍵要素

這裏我們定義 ArticleState 文章狀態,主要包含主題、標題、正文、摘要和插圖 5 個要素:

from typing_extensions import TypedDict
class ArticleState(TypedDict):
    """文章狀態"""
    topic: str  # 主題
    title: str  # 標題
    content: str  # 正文
    summary: str  # 摘要
    image_path: str  # 插圖路徑

接下來,我們進行一些資源準備的工作。本次的項目我們採用 DeepSeek-V3 作爲基礎模型,所以要創建好 DeepSeek 客戶端。這裏我就直接使用 LangChain 封裝好的 ChatOpenAI 類:

import os
import dotenv
from langchain_openai import ChatOpenAI
# 加載環境變量
dotenv.load_dotenv()
def get_deepseek() -> ChatOpenAI:
    """
    獲取DeepSeek客戶端
    :return: DeepSeek客戶端
    """
    return ChatOpenAI(
        model="deepseek-chat",  # 使用DeepSeek-V3 Chat Model
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url=os.getenv("DEEPSEEK_API_BASE"),
    )

然後,我們定義一個全局的 SystemPrompt,在裏面統一文章的主題和行文風格,這邊有利於文章整體結構保持一致:

# 文章的全局System Prompt
SYSTEM_PROMPT = """
你是一位資深的內容編輯,擅長各種題材的寫作。
你正在撰寫一篇主題爲 {topic} 的文章,字數在4000字左右,你力求做到結構嚴謹、行文優美、可讀性強。
特別注意:你所有生成的內容均採用英文。
"""
  1. 標題 Node ==========

狀態定義完成之後,我們就可以來設計 Node 節點了。如果同學忘了 Node 的概念和處理流程,可以複習下前面的課程

按照我們設計的文章撰寫流程,總共需要 4 個 Node:

第一個來實現 TitleNode,它的邏輯非常簡單,就是根據用戶輸入的主題 topic ,創作文章的標題:

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from deepseek import get_deepseek
from prompt import SYSTEM_PROMPT, TITLE_PROMPT
from state import ArticleState
# 標題Prompt
TITLE_PROMPT = """
結合指定的主題,給文章起一個貼切的標題。
特別注意:只生成一個最合適的標題即可,以純文本格式返回,無需生成其他額外信息。
"""
def title_node(state: ArticleState) -> ArticleState:
    """
    文章標題節點
    :param state: 當前文章狀態
    :return: 處理後的文章狀態
    """
    # 從當前狀態中獲取所需信息
    if not state["topic"]:
        raise ValueError("未指定文章主題!")
    topic = state["topic"]
    # 構造prompt
    prompt = ChatPromptTemplate.from_messages([
        ("system", SYSTEM_PROMPT),
        ("user", TITLE_PROMPT)
    ])
    # 獲取DeepSeek客戶端
    deepseek = get_deepseek()
    # 構造Chain
    chain = prompt | deepseek | StrOutputParser()
    # 調用Chain,獲取結果
    title = chain.invoke({"topic": topic})
    # 更新狀態
    state["title"] = title
    print(f"文章標題生成完成:{title}")
    # 返回處理後的狀態
    return state
  1. 正文 Node ==========

第二個是 ContentNode,它根據文章的主題和標題,生成文章的正文內容:

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from deepseek import get_deepseek
from prompt import SYSTEM_PROMPT, CONTENT_PROMPT
from state import ArticleState
# 正文Prompt
CONTENT_PROMPT = """
文章的標題爲:《{title}》
接下來,按照總-分-總的結構,生成文章的主要段落。段落數在3~6段之間,可以結合你的經驗和主題靈活調整。
下面是一個可供參考的段落結構:
1. xxx綜述
2. xxx的歷史演進
3. xxx的核心概念
4. xxx實戰
5. xxx總結
格式要求:生成的內容純文本格式,每一段都要有一個二級標題(##)。段落之間保留2行空行。
"""
def content_node(state: ArticleState) -> ArticleState:
    """
    文章正文內容節點
    :param state: 當前文章狀態
    :return: 處理後的文章狀態
    """
    # 從當前狀態中獲取所需信息
    if not state["topic"]:
        raise ValueError("未指定文章主題!")
    if not state["title"]:
        raise ValueError("文章標題缺失!")
    topic = state["topic"]
    title = state["title"]
    # 構造prompt
    prompt = ChatPromptTemplate.from_messages([
        ("system", SYSTEM_PROMPT),
        ("user", CONTENT_PROMPT)
    ])
    # 獲取DeepSeek客戶端
    deepseek = get_deepseek()
    # 構造Chain
    chain = prompt | deepseek | StrOutputParser()
    # 調用Chain,獲取結果
    content = chain.invoke({"topic": topic, "title": title})
    # 更新狀態
    state["content"] = content
    print(f"文章正文生成完成:共 {len(content)} 字")
    # 返回處理後的狀態
    return state
  1. 摘要 Node ==========

繼續來實現第三個 SummaryNode,基於全文的內容生成摘要總結。之所以需要摘要,是爲了後面的生成插圖做準備。

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from deepseek import get_deepseek
from prompt import SYSTEM_PROMPT, SUMMARY_PROMPT
from state import ArticleState
# 摘要Prompt
SUMMARY_PROMPT = """
現在,文章已經寫好了,請你根據文章的主題、標題和正文內容,生成一段30~60字的摘要,總結這篇文章的核心內容。
標題:{title}
正文內容:{content}
"""
def summary_node(state: ArticleState) -> ArticleState:
    """
    文章摘要節點
    :param state: 當前文章狀態
    :return: 處理後的文章狀態
    """
    # 從當前狀態中獲取所需信息
    if not state["topic"]:
        raise ValueError("未指定文章主題!")
    if not state["title"]:
        raise ValueError("文章標題缺失!")
    if not state["content"]:
        raise ValueError("文章正文內容缺失!")
    topic = state["topic"]
    title = state["title"]
    content = state["content"]
    # 構造prompt
    prompt = ChatPromptTemplate.from_messages([
        ("system", SYSTEM_PROMPT),
        ("user", SUMMARY_PROMPT)
    ])
    # 獲取DeepSeek客戶端
    deepseek = get_deepseek()
    # 構造Chain
    chain = prompt | deepseek | StrOutputParser()
    # 調用Chain,獲取結果
    summary = chain.invoke({"topic": topic, "title": title, "content": content})
    # 更新狀態
    state["summary"] = summary
    print(f"文章摘要生成完成:共 {len(summary)} 字")
    # 返回處理後的狀態
    return state
  1. 插圖 Node ==========

最後一個就是 ImageNode,基於文章的摘要配插圖。圖片生成需要多模態大模型,這裏我們使用豆包的 

seedream-3.0-t2i 這個文生圖模型,它的圖片生成質量還是不錯的。(PS. 豆包大模型可以通過火山引擎來使用,新註冊的賬號還有一定的免費額度)。

生圖成功後,會返回一個可訪問的 url ,但是這個 url 通常是有過期時間的,所以我把圖片 download 下來保存到本地,方便後面合併到文章中。

import os
import dotenv
import requests
from openai import OpenAI
from state import ArticleState
# 生圖Prompt
IMAGE_PROMPT = """
你是一位資深的插畫師,擅長爲各種文章創作精美的插圖。
現在有一篇文章,具體信息如下:
主題:{topic}
標題:{title}
摘要:{summary}
請你爲該文章創造一幅插畫,展示文章所要傳達的核心內容。要求畫質精美、表達力強,且意境深遠,給人無盡回味。
特別注意:圖片中的文字全部採用英文。
"""
# 加載環境變量
dotenv.load_dotenv()
# 創建豆包Seedream客戶端
seedream = OpenAI(
    base_url=os.getenv("VOLCENGINE_API_BASE"),
    api_key=os.environ.get("VOLCENGINE_API_KEY"),
)
def image_node(state: ArticleState) -> ArticleState:
    """
    文章插圖節點
    :param state: 當前文章狀態
    :return: 處理後的文章狀態
    """
    # 從當前狀態中獲取所需信息
    if not state["summary"]:
        raise ValueError("未生成文章摘要!")
    summary = state["summary"]
    # 使用豆包的Seedream模型生成圖片
    response = seedream.images.generate(
        model="doubao-seedream-3-0-t2i-250415",
        prompt=summary,
        size="1024x1024",
        response_format="url"
    )
    if len(response.data) == 0:
        raise ValueError("生成圖片失敗!")
    print(f"文章插圖生成成功!url: {response.data[0].url}")
    # 從url下載圖片,並以png格式保存到本地
    image_path = f"./{state["title"]}.png"
    download_image(response.data[0].url, image_path)
    # 更新狀態
    state["image_path"] = image_path
    print(f"文章插圖生成完成,已保存至:{image_path} ")
    # 返回更新後的狀態
    return state
def download_image(url: str, file_path: str) -> None:
    """
    下載圖片
    :param url: 圖片url
    :param file_path: 圖片保存路徑
    :return: None
    """
    # 下載圖片
    response = requests.get(url)
    with open(file_path, "wb") as f:
        f.write(response.content)
  1. Workflow 編排 ==============

到現在爲止,我們所有需要的 Node 就已經開發完了,最後就是把所有的 Node 串聯成一個 Graph,用於編排我們的 Workflow 工作流。

另外,所有生成的中間結果都保持在 ArticleState 中,因此我實現了一個函數  dump_markdown ,把完整的文章以 Markdown 的格式保存到本地:

from langgraph.constants import START, END
from langgraph.graph import StateGraph
from content_node import content_node
from image_node import image_node
from state import ArticleState
from summary_node import summary_node
from title_node import title_node
def write_article(topic: str) -> ArticleState:
    """
    撰寫文章
    :param topic: 文章主題
    :return: 最終生成的文章狀態
    """
    # 創建StateGraph圖結構
    sg = StateGraph(ArticleState)
    #  添加節點
    sg.add_node("title_node", title_node)
    sg.add_node("content_node", content_node)
    sg.add_node("summary_node", summary_node)
    sg.add_node("image_node", image_node)
    # 添加邊
    sg.add_edge(START, "title_node")
    sg.add_edge("title_node""content_node")
    sg.add_edge("content_node""summary_node")
    sg.add_edge("summary_node""image_node")
    sg.add_edge("image_node", END)
    # 編譯生成圖
    graph = sg.compile()
    # 設置初始狀態
    init_state = ArticleState(
        topic=topic,
        title="",
        content="",
        summary="",
        image_path="",
    )
    # 執行Graph,返回生成的最終狀態
    return graph.invoke(init_state)
def dump_markdown(state: ArticleState) -> None:
    """
    將文章以Markdown格式保存到本地
    :param state: 最終狀態
    :return: None
    """
    title = state["title"]
    with open(f"./{title}.md""w") as f:
        # 寫入標題
        f.write(f"# {state["title"]}\n\n")
        # 寫入正文
        f.write(f"{state["content"]}\n\n")
        # 在末尾插入圖片
        f.write(f"![{title}]({state["image_path"]})\n\n")
    print(f"文章已保存至:{title}.md")
if __name__ == '__main__':
    final_state = write_article("大模型底層的Transformer架構")
    dump_markdown(final_state)
  1. 效果展示 =======

到這裏,所有的代碼就開發完成了。我們以 “大模型底層的 Transformer 架構” 爲主題,生成一篇文章,實際的效果還是很好的:

這一套流程執行下來,一個能寫文章、能配插圖的智能編輯 Agent 就這樣實現了。只要你有一個 idea,指定一個主題,一篇 2000 字左右的、圖文並茂的文章分分鐘信手拈來!

我自己對 LangChain、LangGraph 框架的使用經驗比較豐富,而且作爲 Committer 也參與了一些核心功能的設計與開發。整個過程體驗下來,我最大的感受就是:在大模型時代,最核心的技能,就是對於業務需求的深入理解,以及對於複雜問題的建模與設計。有了這些能力的積累,再結合 LangGraph 這些一流的框架的賦能,Agent 開發簡直易如反掌!

本文由 Readfog 進行 AMP 轉碼,版權歸原作者所有。
來源https://mp.weixin.qq.com/s/KHAlAnQL96mCpQegS70BqQ