基于 LangGraph 的多智能体智能写作系统实现

多智能体架构与智能写作系统

单智能体的能力存在上限，而多智能体的协同能产生巨大的乘数效应。当前主流智能体平台的工作流模式通常仅支持信息的单向流动（输入->输出），无法实现真正的双向甚至多向交流。为了实现多智能体协同，本文选用 LangGraph 作为落地框架，因其具备较强的工程化能力。

系统架构

本系统包含两个核心智能体：

1. 写作智能体 (writer)：负责根据需求、风格和参考资料撰写初稿。
2. 评审智能体 (reviewer)：负责对文章进行评审、打分并提供反馈。

两个智能体经过多轮交互，当文章达到设定目标分数或对话轮次满时停止。该流程模仿人类'写初稿 - 修改 - 定稿'的过程。

工作流程

1. 写作智能体根据输入的写作需求生成初稿。
2. 评审智能体根据评审要求给出意见并打分。
3. 系统判断文章分数是否达标或轮次是否已满，若未达标则继续循环。

该系统对提示词的质量有较高要求，用户需定义清晰的写作和评审风格体系。

核心代码实现

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
from langchain_community.tools import DuckDuckGoSearchRun
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain.prompts import ChatPromptTemplate
from langchain.llms import OpenAI

# 定义共享状态
class WritingState(TypedDict):
    topic: str
    draft: Annotated[str, "writer 生成的文章草稿"]
    review_feedback: Annotated[str, "reviewer 的评审反馈"]
    review_score: Annotated[, ]
    revision_count: 
    final_report: Annotated[, ]


 ():
    ()
    topic = state[]
    review_feedback = state[]
     review_feedback:
        prompt = 
    :
        prompt = 
    llm = OpenAI(model_name=)
    draft = llm(prompt)
    revision_count = state[] + 
     {: draft, : revision_count}

 ():
    ()
    draft = state[]
    prompt = 
    llm = OpenAI(model_name=)
    response = llm(prompt)
    review_score = (response.split()[].strip())
    review_feedback = response.split()[].strip()
     {: review_feedback, : review_score}


 ():
    ()
     state[] >   state[] >= :
         END
    :
         


 ():
    workflow = StateGraph(WritingState)
    workflow.add_node(, writer_node)
    workflow.add_node(, reviewer_node)
    workflow.set_entry_point()
    workflow.add_edge(, )
    workflow.add_conditional_edges(
        , should_continue, {
            : ,
            END: END
        }
    )
     workflow.()


 __name__ == :
    app = create_workflow_graph()
    topic = 
    initial_state = WritingState(
        topic=topic,
        draft=,
        review_feedback=,
        review_score=,
        revision_count=,
        final_report=
    )
    final_state = 
     step  app.stream(initial_state):
         node, values  step.items():
            ()
             node == :
                ()
             node == :
                ()
                ()
        final_state = values
     final_state:
        ()
        (final_state[])