使用 langgraph 生成 3 个节点的图：

from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict

class MyGraphState(TypedDict):
  count: int
  msg: str
  
def counter(state: MyGraphState):
  state["count"] += 1
  state["msg"] = f"函数被调用了 {state['count']} 次"
  return state

workflow = StateGraph(MyGraphState)

workflow.add_node("节点1", counter)
workflow.add_node("节点2", counter)
workflow.add_node("节点3", counter)

workflow.add_edge(START, "节点1")
workflow.add_edge("节点1", "节点2")
workflow.add_edge("节点2", "节点3")

workflow.add_edge("节点3", END)

app = workflow.compile()
app.get_graph().draw_mermaid_png(output_file_path = 'flow.png')
print(app.invoke({"count": 0, "msg":"hello"}))

使用 langgraph 生成一个条件节点：

# chains.py
from dotenv import load_dotenv
from langchain_core.messages import HumanMessage
import os
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai import ChatOpenAI

load_dotenv()

reflection_prompt = ChatPromptTemplate.from_messages(
  [
    (
      "system",
      "你是一位有影响力的微博博主，正在对一条微博打分。对这条微博的优缺点进行评判"
      "请提供一个详细的评判，包括长度，热度，风格等等"
    ),
    MessagesPlaceholder(variable_name='messages')
  ]
)

generation_prompt = ChatPromptTemplate.from_messages(
  [
    (
      "system",
      "你是一位微博科技类网红助手，任务是撰写优质的推文帖子。"
      "根据用户的请求，生成尽可能最佳的推文帖子。"
      "如果用户提供修改意见，请根据之前的草稿重新生成修改版。"
    ),
    MessagesPlaceholder(variable_name='messages')
   ]
)

llm = ChatOpenAI(
            verbose=True,
            temperature=0,
            model_name="deepseek-r1",
            api_key=os.environ["OPENAI_API_KEY"],
            base_url=os.environ["OPENAI_ENDPOINT"]
      )

generation_chain = generation_prompt | llm
reflection_chain = reflection_prompt | llm

# main.py
from typing import Sequence, List
from dotenv import load_dotenv

load_dotenv()

from langchain_core.messages import HumanMessage
from langgraph.graph import END, StateGraph, MessagesState

from chains import generation_chain, reflection_chain

GENERATE = 'generate'
REFLECT = 'reflect'

def generation_node(state: MessagesState):
  response = generation_chain.invoke(state['messages'])
  return {'messages': response}

def reflection_node(state: MessagesState):
  response = reflection_chain.invoke(state['messages'])
  return {'messages': response}

builder = StateGraph(MessagesState)
builder.add_node(GENERATE, generation_node)
builder.add_node(REFLECT, reflection_node)

builder.set_entry_point(GENERATE)

def should_continue(state: MessagesState):
  if len(state['messages']) > 3:
    return END
  
  return REFLECT

builder.add_conditional_edges(GENERATE, should_continue, {
  END: END,
  REFLECT: REFLECT
})
builder.add_edge(REFLECT, GENERATE)

graph = builder.compile()
graph.get_graph().draw_mermaid_png(output_file_path = 'flow.png')

if __name__ == '__main__':
  inputs = HumanMessage(content="""改一下这一篇推文让它变得更好:"
                                    来自 LangChain 团队完全开源的一个深度研究 Agent：Open Deep Research。

支持多种模型提供商和搜索工具，能够自行执行复杂的研究任务，从信息收集到分析整理自动化处理。

GitHub：github.com/langchain-ai/open_deep_research

主要功能：

- 支持 OpenAI、Anthropic、Google 等多种模型提供商
- 集成 Tavily、原生网页搜索等多种搜索工具
- 可配置的并发研究单元，提升研究效率
- 支持 MCP 服务器扩展，增强研究能力
- 提供 LangGraph Studio 可视化界面
- 内置评估系统，确保研究质量

提供 LangGraph 服务器本地部署方式，也可部署到 LangGraph Platform 平台使用。""")
  
  initial_state = {
      "messages": [inputs],
  }
  
  # 生成
  # print(generation_chain.invoke(initial_state))
  print(reflection_chain.invoke(initial_state))