最先端の構築に基づく LangGraph 工作流入门 – lyshark – ブログ
従来の LangChain は、抽象コアとして Chain と Runnable を備えた LLM アプリケーション コンポーネントのツールボックスで、プロンプト ワード テンプレート、文書処理、ベクトル取得、ツール呼び出し、メモリ容量などの標準化されたコンポーネントを提供します。その目的は、単純な線形 LLM タスクを迅速に構築することです。 Graph は、ステートフルで動的に反復されるインテリジェントなワークフローの微細な制御に焦点を当てた、実行時の複雑なシナリオのステート チャートです。これはノード、エッジ、およびグローバル状態に基づいており、リング トポロジをサポートしています。これにより、自律的な意思決定と循環的な反復により、複雑な実稼働レベルのビジネス シナリオに効率的に適応できます。
従来の LangChain は、抽象コアとして Chain と Runnable を備えた LLM アプリケーション コンポーネントのツールボックスで、プロンプト ワード テンプレート、文書処理、ベクトル取得、ツール呼び出し、メモリ容量などの標準化されたコンポーネントを提供します。その目的は、単純な線形 LLM タスクを迅速に構築することです。 Graph は、ステートフルで動的に反復されるインテリジェントなワークフローの微細な制御に焦点を当てた、実行時の複雑なシナリオのステート チャートです。これはノード、エッジ、およびグローバル状態に基づいており、リング トポロジをサポートしています。これにより、自律的な意思決定と循環的な反復により、複雑な実稼働レベルのビジネス シナリオに効率的に適応できます。
チェーン ダイアグラムは LangGraph の最も基本的なトポロジであり、単一の線形実行パスのみが存在し、ノードは事前に定義された順序に従って上から下に流れ、状態データを渡します。プロセスには判断条件、分岐ジャンプ、ループバック ロジックがなく、固定ステップ、シンプルなロジック、およびタスク シーンの標準化に適しています。 LangGrap では、ノードを追加し、エッジ関係を構成することで、完全な線形チェーンが実行されます。
単線チェーンコール
単一の直線パスのみがあり、データはノードから下位に送信され、判断、分岐、ループはなく、固定ステップ、単純な論理標準化タスクに適しています。
全体的な構造のフローチャート:
参照コード構造:
from langgraph.graph.message import MessagesState
from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
def qwen_agent_node(state: MessagesState):
llm = ChatOpenAI(
model="qwen2.5-1.5b-instruct-q4_k_m.gguf",
base_url="
api_key="dummy",
temperature=0.7,
max_tokens=512,
)
resp = llm.invoke(state["messages"])
return {"messages": [resp]}
if __name__ == "__main__":
# 开始节点
builder = StateGraph[MessagesState, None, MessagesState, MessagesState](MessagesState)
# 增加节点
builder.add_node("qwen_agent", qwen_agent_node)
# 构建START起始边
builder.add_edge(START, "qwen_agent")
# 构建END结束边
builder.add_edge("qwen_agent", END)
# 编译工作流
graph = builder.compile()
# 调用LLM
result = graph.invoke({
"messages": [
SystemMessage(content="你是简短回答助手，你的名字是小张"),
HumanMessage(content="你好，请简单介绍一下自己？")
]
})
for msg in result["messages"]:
print(f"{msg.type}: {msg.content}")
# pygraphviz 查询图并绘制流程图
png_bytes = graph.get_graph().draw_mermaid_png()
with open("agent_graph.png", "wb") as f:
f.write(png_bytes)
実行するコードは次の情報を出力します。
system:
你是简短回答助手，你的名字是小张
human:
你好，请简单介绍一下自己？
ai:
我的主要任务是快速、准确地回答用户的问题。如果您有任何问题，欢迎随时提问。
電影词鼕式引说
正词链式歌词词词词词これはLLM呼び出し、出力結果のすべてを承知、複雑なタスクを分断し、単一の単純なノードを多段階の処理プロセスにアップグレードし、エージェントに複雑なタスクを逆アセンブルして分単位で実行できるようにします。
ワードリンク型呼び出しのプロンプトを実現するために、元の単一の qwen_agent を機能の異なる 2 つに分割します。
- Planner (プランナー): ユーザーの責任ある分析 意图、安全安全电影戦略
- ライター (宣写者): プランナーの戦略に従って、最終応答を生成
全体的な構造のフローチャート:
参照コード構造:
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from typing import TypedDict, Annotated
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
# 自定义状态：messages 保存用户对话
class State(TypedDict):
messages: Annotated[list, add_messages]
plan: str
def get_llm():
return ChatOpenAI(
model="qwen2.5-1.5b-instruct-q4_k_m.gguf",
base_url="
api_key="dummy",
temperature=0.7,
max_tokens=512,
)
# 定义节点：规划者 (Planner)
def planner_node(state: State):
llm = get_llm()
system_prompt = SystemMessage(content="""
任务：仅分析用户意图，生成简短执行计划。
禁止直接回答用户问题！禁止自我介绍！
输出格式：
用户意图：xxx
回答策略：xxx
只输出这两行，不要额外内容。
""")
resp = llm.invoke([system_prompt] + state["messages"])
# 把规划结果存入plan字段
return {"plan": resp.content}
# 定义节点：撰写者 (Writer)
def writer_node(state: State):
llm = get_llm()
# writer拿到plan文本，作为参考
system_prompt = SystemMessage(content="""
你叫小张，简短友好回答用户。
参考下面的【执行计划】，面向用户直接给出回答。
不要复述计划内容，不要提到规划者。
""")
# 构造消息：系统提示 + 用户原始提问 + 把plan当成一段参考文本
user_msg = state["messages"][0].content
plan_text = state["plan"]
prompt_messages = [
system_prompt,
HumanMessage(content=f"用户问题：{user_msg}\n【执行计划】{plan_text}")
]
resp = llm.invoke(prompt_messages)
return {"messages": [resp]}
if __name__ == "__main__":
# 构建工作流
builder = StateGraph(State)
# 添加节点
builder.add_node("planner", planner_node)
builder.add_node("writer", writer_node)
# 构建链式流程 (Chain)
# 流程：开始 -> 规划者 -> 撰写者 -> 结束
builder.add_edge(START, "planner")
builder.add_edge("planner", "writer")
builder.add_edge("writer", END)
# 编译工作流
graph = builder.compile()
# 调用LLM执行链式调用
print("--- 开始执行链式调用 ---")
result = graph.invoke({
"messages": [HumanMessage(content="你好，请简单介绍一下你自己？")],
"plan": ""
})
print("\n--- 状态输出 ---")
print(f"Planner规划内容：\n{result['plan']}\n")
print(f"最终回答(writer输出)：\n{result['messages'][-1].content}")
実行するコードは次の情報を出力します。
--- 开始执行链式调用 ---
--- 状态输出 ---
Planner规划内容：
用户意图：了解机器人自身的信息
回答策略：机器人是AI助手，我是一个能够回答问题、提供信息和执行任务的程序。
最终回答(writer输出)：
你好，我叫小张，我是一个能够回答问题、提供信息和执行任务的程序。
流程图已保存为 chain_agent_graph.png
上行化铝式アニメーション生機
マルチエッジ構成により、ノードを並列実行できます。つまり、古典的なファンアウト/ファンイン (扇出/扇入) モードは、古典的なファンイン LLM です。複数の独立したタスクを同時に実行し、タスクの処理速度を向上させます。
論理実行コア:
- START ノードは複数の LLM ノードを同時にトリガーし、実行します
- 金属電影校閉正視全可行行设计安全敯
- すべてのタスクが完了したら、ノード集約アグリゲーターに入り、最終結果を出力します。
全体的な構造のフローチャート:
参照コード構造:
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
# 定义图状态结构体
class State(TypedDict):
topic: str # 主题，用户输入
joke: str # 生成的笑话
story: str # 生成的故事
poem: str # 生成的诗歌
combined_output: str# 聚合后的最终输出文本
def get_llm():
return ChatOpenAI(
model="qwen2.5-1.5b-instruct-q4_k_m.gguf",
base_url="
api_key="dummy",
temperature=0.7,
max_tokens=512,
)
def call_llm_1(state: State):
"""节点1：调用LLM生成笑话"""
llm = get_llm()
msg = llm.invoke(f"写一个关于{state['topic']}的笑话")
return {"joke": msg.content}
def call_llm_2(state: State):
"""节点2：调用LLM生成小故事"""
llm = get_llm()
msg = llm.invoke(f"写一个关于{state['topic']}的短篇小故事")
return {"story": msg.content}
def call_llm_3(state: State):
"""节点3：调用LLM生成短诗"""
llm = get_llm()
msg = llm.invoke(f"写一首关于{state['topic']}的短诗")
return {"poem": msg.content}
def aggregator(state: State):
"""聚合节点：把笑话、故事、诗歌合并成一段完整输出"""
combined = f"这是关于[{state['topic']}]的故事、笑话和诗歌！\n\n"
combined += f"故事：\n{state['story']}\n\n"
combined += f"笑话：\n{state['joke']}\n\n"
combined += f"诗歌：\n{state['poem']}"
return {"combined_output": combined}
if __name__ == "__main__":
# 创建状态图实例
parallel_builder = StateGraph(State)
# 添加所有节点到图中
parallel_builder.add_node("call_llm_1", call_llm_1)
parallel_builder.add_node("call_llm_2", call_llm_2)
parallel_builder.add_node("call_llm_3", call_llm_3)
parallel_builder.add_node("aggregator", aggregator)
# 添加边关系：扇出 Fan-out
# 流程起点START，同时触发三个LLM节点（逻辑并行）
parallel_builder.add_edge(START, "call_llm_1")
parallel_builder.add_edge(START, "call_llm_2")
parallel_builder.add_edge(START, "call_llm_3")
# 扇入 Fan-in：三个LLM节点全部执行完成后，才进入聚合节点
parallel_builder.add_edge("call_llm_1", "aggregator")
parallel_builder.add_edge("call_llm_2", "aggregator")
parallel_builder.add_edge("call_llm_3", "aggregator")
# 聚合完成后，流程结束
parallel_builder.add_edge("aggregator", END)
# 编译工作流
parallel_workflow = parallel_builder.compile()
# 执行工作流
result = parallel_workflow.invoke({"topic": "小猫"})
# 打印合并后的结果
print(result["combined_output"])
実行するコードは次の情報を出力します。
这是关于[小猫]的故事、笑话和诗歌！
故事：
在一个宁静的小镇上，住着一只名叫米咪的小猫。米咪是一只非常好奇、活泼的小猫，它总是喜欢探索周围的世界。一天，米咪决定去探索小镇的另一个角落，那是一个被遗忘的花园。
在花园中，米咪遇到了一只正在玩耍的蝴蝶。蝴蝶告诉米咪，花园里藏着一个秘密，一个隐藏的洞穴。米咪被这个神秘的洞穴吸引，决定去探索。
它穿过一片茂密的树林，终于来到了洞穴的入口。洞穴内部光线昏暗，但米咪并不害怕，它勇敢地走进了黑暗的洞穴。洞穴里有一片温暖的阳光，它发现了一只可爱的小兔子。
小兔子对米咪说：“欢迎来到我的家，我的名字叫做小白。”小白邀请米咪和它一起探索这个洞穴的秘密。它们发现了一个隐藏的水井，还发现了一片美味的野果。
在洞穴里度过了愉快的时光后，米咪和小白又回到了阳光下。小白告诉米咪：“现在，你已经知道这个花园的秘密了。你是一个勇敢的小猫，我非常高兴能与你一起度过这段时光。”
米咪感激地离开了洞穴，带着对这个神秘花园的深深爱意和对小兔子小白的感激之情。从此，米咪成了小镇上的小英雄，因为它的勇敢和好奇心，它赢得了所有人的喜爱。而那个神秘的花园，也成为了米咪心中最美好的回忆。
笑话：
小猫，一只小猫，它没有尾巴，它有几只耳朵？（答案：两只）
诗歌：
小猫的轻笑
在阳光下跳跃，
小爪轻轻敲打着地面，
它的眼眸闪烁着好奇，
寻找着世界的秘密。
在雨中奔跑，
雨珠在它的小尾巴上滚动，
它享受着雨水带来的清凉，
在小池塘里嬉戏着。
在夜晚静悄悄，
它蜷缩在温暖的角落，
听风声，听雨声，
感受着世界的宁静与美好。
分岐判断は、意図認識とルーティング メカニズムに依存して自律的な意思決定を実現するインテリジェンスの中心的な機能であり、ワークフローはユーザー入力コンテンツに基づいて異なるビジネス処理ノードに自動的に分散され、差別化されたロジックを実行できます。この種のルーティング ワークフローは、インテリジェントな顧客サービス、タスク分類、ユーザー意図認識などのシナリオで広く使用されている入力分類とタスク分散を完了できます。さまざまな種類のリクエストに対して、専用の処理リンクを有効にすることができます。たとえば、顧客サービス システムを相談タイプに分割し、価格相談、返金申請、および販売後の質問を対応する処理ロジックにルーティングできます。
発行小意图通记電影
基本的な意図ルーティングスキームは、意図判定を完了するために大規模なモデルのフリーテキスト出力を採用し、意四認識ノードによってユーザーの質問を読み取り、分類ラベルを出力し、ルーティング機能を通じてラベル結果を読み取り、使用しますadd_conditional_edges条件エッジを登録し、分岐ジャンプを実現します。この場合、ユーザーは数学的な計算と日常のチャットに分かれます。
全体的な構造のフローチャート:
参照コード構造:
from langgraph.graph.message import MessagesState
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
# 本地调用大模型接口
def get_llm():
return ChatOpenAI(
model="qwen2.5-1.5b-instruct-q4_k_m.gguf",
base_url="
api_key="dummy",
temperature=0.7,
max_tokens=512,
)
# 意图识别节点
def intent_detect_node(state: MessagesState):
"""意图识别节点：判断用户是数学问题还是闲聊，输出到messages，后面路由读取该判断"""
llm = get_llm()
prompt = SystemMessage(
content="""你是意图分类器。
请分析用户问题，只输出一个单词，只能二选一：math 或者 chat。
math：数学计算类问题；chat：日常闲聊对话。禁止输出多余解释。
"""
)
resp = llm.invoke([prompt] + state["messages"])
return {"messages": [resp]}
# 数学处理节点
def math_handler_node(state: MessagesState):
"""处理数学问题分支"""
llm = get_llm()
sys_msg = SystemMessage(content="你是数学助手，简洁解答数学题目。")
resp = llm.invoke([sys_msg] + state["messages"])
return {"messages": [resp]}
# 闲聊处理节点
def chat_handler_node(state: MessagesState):
"""处理闲聊分支"""
llm = get_llm()
sys_msg = SystemMessage(content="你是闲聊助手，简短友好聊天。")
resp = llm.invoke([sys_msg] + state["messages"])
return {"messages": [resp]}
# 路由函数
def route_intent(state: MessagesState):
"""条件分支路由函数：读取上一步意图识别结果，返回下一个节点名字"""
last_msg = state["messages"][-1]
content = last_msg.content.strip().lower()
if "math" in content:
return "math_handler_router"
else:
return "chat_handler_router"
if __name__ == "__main__":
# 构建状态图
builder = StateGraph(MessagesState)
# 添加三个节点：意图识别、数学处理、闲聊处理
builder.add_node("intent_detect", intent_detect_node)
builder.add_node("math_handler", math_handler_node)
builder.add_node("chat_handler", chat_handler_node)
# 起始边：START -> 意图识别节点
builder.add_edge(START, "intent_detect")
# 条件分支：通过intent_detect输出后交给route_intent做跳转
builder.add_conditional_edges(
source="intent_detect",
path=route_intent,
path_map={
"math_handler_router": "math_handler",
"chat_handler_router": "chat_handler"
}
)
# math_handler 和 chat_handler 两个分支都结束到END
builder.add_edge("math_handler", END)
builder.add_edge("chat_handler", END)
# 编译图
graph = builder.compile()
print("------ 数学问题,会走math分支 ------")
result1 = graph.invoke({
"messages": [
HumanMessage(content="35乘以12等于多少？")
]
})
for msg in result1["messages"]:
print(f"{msg.type}: {msg.content}")
print("\n------ 闲聊问题,会走chat分支 ------")
result2 = graph.invoke({
"messages": [
HumanMessage(content="你好，知道Python吗？")
]
})
for msg in result2["messages"]:
print(f"{msg.type}: {msg.content}")
実行するコードは次の情報を出力します。
------ 数学问题,会走math分支 ------
human: 35乘以12等于多少？
ai: math
ai: math(35 * 12) = 420
------ 闲聊问题,会走chat分支 ------
human: 你好，知道Python吗？
ai: chat
ai: 可以聊天，但是不会做Python，不过你可以试试看。
正確な出力構造
不安定な欠陥のテキスト分析の基礎を解決するために、このケースが採用されますwith_structured_outputPydantic モデル制約、大規模なモデル出力形式により、LLM はルート ソースから JSON 構造の標準化に戻り、テキスト分析によって引き起こされる判断エラーを排除し、ルーティングの安定性と精度を向上させます。
事前定義されたケースRouteモデル限定の分類結果のみ使用可能poem、story、joke三者一. プロセスの実行。ユーザー入力は最初にルーティング ノードに送信され、Pydantic モデル仕様に従って LLM が構造化された決定結果を出力して状態を保存します。ルーティング関数は状態内の決定フィールドを読み取り、タスクを 3 つの独立したノードの 1 つに自動的に分割し、ノードは対応するコンテンツを生成してプロセスを直接終了します。
全体的な構造のフローチャート:
参照コード構造:
from typing import TypedDict
from typing_extensions import Literal
from pydantic import BaseModel, Field
from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(
model="qwen2.5-1.5b-instruct-q4_k_m.gguf",
base_url="
api_key="dummy",
temperature=0.7,
max_tokens=512,
)
# Pydantic模型：定义结构化输出，用于路由判断
class Route(BaseModel):
step: Literal["poem", "story", "joke"] = Field(
None, description="路由的下一步，只能选 poem / story / joke"
)
# 给LLM绑定结构化输出能力，强制输出符合Route模型的结果
router = llm.with_structured_output(Route)
# 定义图状态，保存流转数据
class State(TypedDict):
input: str # 用户原始输入
decision: str # 路由决策结果：poem / story / joke
output: str # 最终生成内容
# ---------------------- 节点定义 ----------------------
def llm_call_1(state: State):
"""节点1：写故事"""
result = llm.invoke(f"写一个短篇故事：{state['input']}")
return {"output": result.content}
def llm_call_2(state: State):
"""节点2：写笑话"""
result = llm.invoke(f"写一个笑话：{state['input']}")
return {"output": result.content}
def llm_call_3(state: State):
"""节点3：写诗歌"""
result = llm.invoke(f"写一首短诗：{state['input']}")
return {"output": result.content}
def llm_call_router(state: State):
"""路由节点：使用LLM判断用户需求，输出路由决策"""
decision = router.invoke(
[
SystemMessage(
content="根据用户请求，判断用户想要故事(story)、笑话(joke)还是诗歌(poem)，只返回对应的分类。"
),
HumanMessage(content=state["input"]),
]
)
# 将决策存入state
return {"decision": decision.step}
# 条件路由函数：根据decision的值，返回下一个节点名称
def route_decision(state: State):
if state["decision"] == "story":
return "llm_call_1"
elif state["decision"] == "joke":
return "llm_call_2"
elif state["decision"] == "poem":
return "llm_call_3"
if __name__ == "__main__":
# 构建状态图
router_builder = StateGraph(State)
# 添加所有节点
router_builder.add_node("llm_call_1", llm_call_1)
router_builder.add_node("llm_call_2", llm_call_2)
router_builder.add_node("llm_call_3", llm_call_3)
router_builder.add_node("llm_call_router", llm_call_router)
# 边配置：起点进入路由节点
router_builder.add_edge(START, "llm_call_router")
# 添加条件边：路由节点执行完后，根据route_decision函数跳转到对应分支
router_builder.add_conditional_edges(
"llm_call_router", # 来源节点
route_decision, # 判断函数
{
"llm_call_1": "llm_call_1",
"llm_call_2": "llm_call_2",
"llm_call_3": "llm_call_3",
},
)
# 三个生成节点执行完毕直接结束流程
router_builder.add_edge("llm_call_1", END)
router_builder.add_edge("llm_call_2", END)
router_builder.add_edge("llm_call_3", END)
# 编译图
router_workflow = router_builder.compile()
# 测试运行
user_input = "帮我写一个关于小狗的故事"
result = router_workflow.invoke({"input": user_input})
print(f"用户输入：{result['input']}")
print(f"路由判断：{result['decision']}")
print(f"生成结果：{result['output']}")
実行するコードは次の情報を出力します。
用户输入：
帮我写一个关于小狗的故事
路由判断：
story
生成结果：
在一个宁静的小镇上，住着一只名叫豆豆的小狗。豆豆是一只非常聪明的狗，它拥有着敏锐的嗅觉和敏捷的身手。在小镇上，豆豆不仅是邻居们的宠儿，更是孩子们的玩伴。
从那以后，豆豆成了小镇的英雄，它不仅因为它的忠诚和智慧赢得了人们的喜爱，也因为它的勇敢和善良，成为了大家心中的保护神。而豆豆自己也更加了解了自己，它明白了，真正的力量来自于内心的善良和勇敢。
終了条件が満たされるまで、線形リンク、分岐ルートは 1 回限りのタスク処理のみを実行でき、リアル インテリジェンスは 1 回限りのタスク処理のみを実行でき、リアル インテリジェンスは 1 回限りのタスク処理のみを実行でき、リアル インテリジェンスは 1 回限りのタスク処理のみを実行できます。
本使用可能 ReAct 思想恭建電視体，使用 with_structured_output モデルに固定構造を出力させる ReActResponse、思考内容が含まれます thought 和设计 actionアクションは次の 2 つだけをサポートします：continue 例文帳に追加尚不可，电影电影下載；answer 十分な情報を表し、タスクを終了できます。ステータスは、メッセージのリスト、現在のサイクル数、最大サイクル制限、および思考記録の各ラウンド、完全な反復トラックを維持します。
全体的な構造のフローチャート:
参照コード構造:
from typing import Annotated, TypedDict, Literal
from langgraph.graph.message import add_messages
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage
from pydantic import BaseModel
# 定义模型强制输出的JSON结构
class ReActResponse(BaseModel):
thought: str # 思考/回答正文
action: Literal["continue", "answer"] # 动作二选一
# 定义状态结构
class ReActState(TypedDict):
messages: Annotated[list, add_messages]
loop_count: int
max_loop: int
round_records: list[dict]
def get_llm():
base_llm = ChatOpenAI(
model="qwen2.5-1.5b-instruct-q4_k_m.gguf",
base_url="
api_key="dummy",
temperature=0.1,
max_tokens=1024,
)
# 强制结构化输出
llm = base_llm.with_structured_output(ReActResponse)
return llm
# 定义思考节点
def react_think_node(state: ReActState):
llm = get_llm()
sys_prompt = SystemMessage(content="""
你是ReAct思考助手，分步分析用户问题。
请输出思考内容，同时选择动作：
- continue：分析不完整，需要继续补充思考
- answer：分析完整，可以结束任务
""")
# 调用模型，直接得到解析好的对象
result: ReActResponse = llm.invoke([sys_prompt] + state["messages"])
new_loop_count = state["loop_count"] + 1
# 把思考正文存入AI消息，留给后续链路查看历史
ai_msg = AIMessage(content=result.thought)
# 获取最近用户提问
last_user_q = ""
for m in reversed(state["messages"]):
if isinstance(m, HumanMessage):
last_user_q = m.content
break
record = {
"round": new_loop_count,
"question": last_user_q,
"thought": result.thought,
"action": result.action
}
old_records = state.get("round_records", [])
print(f"\n[第 {new_loop_count} 轮思考]")
print(f"action={result.action}")
print(f"thought:\n{result.thought}")
return {
"messages": [ai_msg],
"loop_count": new_loop_count,
"round_records": old_records + [record]
}
def react_router(state: ReActState) -> Literal["react_think_router", "continue_prompt_router", "end_router"]:
current_loop = state["loop_count"]
max_loop = state["max_loop"]
last_record = state["round_records"][-1]
action = last_record["action"]
# 最大循环保护
if current_loop >= max_loop:
print(f">>> 达到最大循环 {max_loop}，强制结束任务")
return "end_router"
if action == "answer":
print(">>>识别 action=answer，任务完成结束")
return "end_router"
elif action == "continue":
print(">>>识别 action=continue，继续思考")
return "continue_prompt_router"
else:
print(">>>未知action，直接结束")
return "end_router"
def continue_prompt_node(state: ReActState):
"""追加新的human消息，保证消息列表以user结尾，再回到思考节点"""
prompt = HumanMessage(content="请基于前面全部思考，继续补充完善该问题的分析。")
return {
"messages": [prompt]
}
if __name__ == "__main__":
# 构建状态图
builder = StateGraph(ReActState)
# 添加所有节点
builder.add_node("react_think", react_think_node)
builder.add_node("continue_prompt", continue_prompt_node)
# 边配置：起点进入思考节点
builder.add_edge(START, "react_think")
# 边配置：思考节点根据路由判断，继续思考或继续补充思考
builder.add_conditional_edges(
source="react_think",
path=react_router,
path_map={
"react_think_router": "react_think",
"continue_prompt_router": "continue_prompt",
"end_router": END
}
)
# 边配置：继续补充思考节点，再回到思考节点
builder.add_edge("continue_prompt", "react_think")
# 编译图
graph = builder.compile()
print("结构化输出版本运行")
result = graph.invoke({
"messages": [
HumanMessage(content="请分点解释什么是微服务架构，说明优缺点。")
],
"loop_count": 0,
"max_loop": 3,
"round_records": []
})
print("全部轮次汇总")
for r in result["round_records"]:
print(f"---轮次 {r['round']} ---")
print(f"question：{r['question']}")
print(f"action：{r['action']}")
print(f"thought：\n{r['thought']}")
print(f"实际循环次数：{result['loop_count']}")
実行するコードは次の情報を出力します。
结构化输出版本运行
[第 1 轮思考]
action=continue
thought:
微服务架构是一种将应用程序分解为多个小型、独立的、可管理的组件的技术。每个微服务都是一个独立的进程或服务，可以运行在不同的服务器上，可以独立部署和扩展。
>>>识别 action=continue，继续思考
[第 2 轮思考]
action=continue
thought:
微服务架构的优点和缺点已经分析完毕，现在需要继续补充完善该问题的分析。
>>>识别 action=continue，继续思考
[第 3 轮思考]
action=continue
thought:
微服务架构的优点和缺点已经分析完毕，现在需要继续补充完善该问题的分析。
>>> 达到最大循环 3，强制结束任务
全部轮次汇总
---轮次 1 ---
question：请分点解释什么是微服务架构，说明优缺点。
action：continue
thought：
微服务架构是一种将应用程序分解为多个小型、独立的、可管理的组件的技术。每个微服务都是一个独立的进程或服务，可以运行在不同的服务器上，可以独立部署和扩展。
---轮次 2 ---
question：请基于前面全部思考，继续补充完善该问题的分析。
action：continue
thought：
微服务架构的优点和缺点已经分析完毕，现在需要继续补充完善该问题的分析。
---轮次 3 ---
question：请基于前面全部思考，继续补充完善该问题的分析。
action：continue
thought：
微服务架构的优点和缺点已经分析完毕，现在需要继续补充完善该问题的分析。
实际循环次数：3
以前紹介したチェーン図、分岐判定図、ループ判定図は、タスクの分岐数があらかじめ決まっていることがほとんどです。ただし、実際のビジネス シナリオの多くでは、サブタスクの数を事前に決定することはできず、入力に基づいて大規模なモデルによってタスクを動的に分解する必要があります。
核となる考えは次の 3 つの部分に分かれています。
- 编排器（オーケストレーター）： マスター制御ノードとして、LLM を呼び出して元のタスクを逆アセンブルし、複数のサブタスクを生成します。
- 工作器（ワーカー）：排器下発行のサブタスクを受け取り、独立した並列実行；
- 汇总器（シンセサイザー）: すべてのタスクが実行されるまで待ち、すべての出力サブタスクを収集し、それらを最終出力に統合します。
最大のモデルは LangGraph です Send API、動的にオペレーターノードを作成します。事前にブランチの数を記録する必要はありません。フィルターを使用してサブタスクの数を出力すると、対応する数の並列ワーカーが自動的に生成されます。 各ワーカーは独立したローカル状態を持ち、ワーカーのすべての実行結果はグローバル共有状態に書き込まれます。completed_sections、 使用operator.addリストが自動的にマージされることを実現し、共有ノードは共有ステータスを読み取ることができます。
全体的な構造のフローチャート:
参照コード構造:
from typing import Annotated, List, TypedDict
import operator
from typing_extensions import Literal
from pydantic import BaseModel, Field
from langgraph.graph import StateGraph, START, END
from langgraph.types import Send
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
llm = ChatOpenAI(
model="qwen2.5-1.5b-instruct-q4_k_m.gguf",
base_url="
api_key="dummy",
temperature=0.1,
max_tokens=1024,
)
# 报告的单个章节
class Section(BaseModel):
name: str = Field(
description="报告这个章节的名称",
)
description: str = Field(
description="简要描述本章要覆盖的主题和内容要点",
)
# 包含全部章节列表
class Sections(BaseModel):
sections: List[Section] = Field(
description="报告的所有章节列表",
)
# 将LLM绑定结构化输出，强制返回Sections结构
planner = llm.with_structured_output(Sections)
# 全局图状态
class State(TypedDict):
topic: str # 报告主题，用户输入
sections: list[Section] # 编排器生成的章节大纲列表
completed_sections: Annotated[
list, operator.add
] # 所有worker并行写入这个字段，operator.add代表列表追加合并
final_report: str # 最终合并完成的完整报告
# Worker子任务状态（每个独立worker的状态）
class WorkerState(TypedDict):
section: Section # 当前worker负责的章节对象
completed_sections: Annotated[list, operator.add] # 写完的章节内容，会合并回全局state
# 节点定义
def orchestrator(state: State):
"""编排器节点：生成报告整体大纲，输出多个章节规划"""
report_sections = planner.invoke(
[
SystemMessage(content="为用户生成一份报告的章节规划。"),
HumanMessage(content=f"报告主题：{state['topic']}"),
]
)
return {"sections": report_sections.sections}
def llm_call(state: WorkerState):
"""工作器节点：单个worker，负责撰写一个章节内容"""
section = llm.invoke(
[
SystemMessage(
content="按照提供的章节名称和描述撰写报告章节。不要多余开场白，使用markdown格式。"
),
HumanMessage(
content=f"章节名称：{state['section'].name}，章节描述：{state['section'].description}"
),
]
)
# 返回章节文本，会自动合并到 completed_sections
return {"completed_sections": [section.content]}
def synthesizer(state: State):
"""汇总节点：把所有已经写完的章节拼接成完整报告"""
completed_sections = state["completed_sections"]
# 使用分割线分隔各个章节
completed_report_sections = "\n\n---\n\n".join(completed_sections)
return {"final_report": completed_report_sections}
def assign_workers(state: State):
"""条件路由函数：为每一个章节，动态派发一个worker任务，使用Send API并行执行"""
# Send(节点名，传递给该节点的状态字典)
return [Send("llm_call", {"section": s}) for s in state["sections"]]
if __name__ == "__main__":
# 构建状态图
orchestrator_worker_builder = StateGraph(State)
# 添加节点
orchestrator_worker_builder.add_node("orchestrator", orchestrator)
orchestrator_worker_builder.add_node("llm_call", llm_call)
orchestrator_worker_builder.add_node("synthesizer", synthesizer)
# 添加边
orchestrator_worker_builder.add_edge(START, "orchestrator")
# 编排器执行完成后，调用assign_workers批量派发并行worker任务
orchestrator_worker_builder.add_conditional_edges(
"orchestrator", assign_workers, ["llm_call"]
)
# 所有worker全部执行完毕后，进入汇总节点
orchestrator_worker_builder.add_edge("llm_call", "synthesizer")
orchestrator_worker_builder.add_edge("synthesizer", END)
# 编译图
orchestrator_worker = orchestrator_worker_builder.compile()
# 开始测试运行
result = orchestrator_worker.invoke({"topic": "微服务架构介绍"})
print("最终完整报告")
print(result["final_report"])
実行するコードは次の情報を出力します。
最终完整报告
微服务架构（Microservices Architecture）是一种软件架构风格，它将一个大型的应用程序分解成多个小型、独立的、功能单一的服务。每个服务都具有独立的代码库、数据库和部署环境，这些服务通过轻量级的通信协议（
如HTTP、RESTful API或消息队列）进行交互。微服务架构的核心思想是通过将应用程序分解成多个小而独立的服务来提高系统的可扩展性、可维护性和灵活性。
前述のループ パターンの例は、知的体が反復的に思考し続け、評価ツールと最適化モードが生成と評価の 2 つの独立した役割に分かれています。用電：最適化器（電影器）が出力内容を担当し、評価器は事前に設定された基準に従って出力に専念し、改善提案を生成します。評価の結果、内容が規格に達していないと判断された場合は、フィードバック情報を何度もジェネレータに持ち帰って、評価が規格に達するとそのまま処理を終了します。
全体的な構造のフローチャート:
参照コード構造:
from typing import TypedDict, Literal
from pydantic import BaseModel, Field
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(
model="qwen2.5-1.5b-instruct-q4_k_m.gguf",
base_url="
api_key="dummy",
temperature=0.1,
max_tokens=1024,
)
# Pydantic结构化输出模型：笑话评估结果
class Feedback(BaseModel):
grade: Literal["funny", "not funny"] = Field(
description="判断笑话好笑(funny)还是不好笑(not funny)"
)
feedback: str = Field(
description="如果笑话不好笑，给出改进的反馈意见。"
)
# 绑定结构化输出，评估器会返回Feedback对象
evaluator = llm.with_structured_output(Feedback)
# 图状态定义
class State(TypedDict):
joke: str # 生成的笑话文本
topic: str # 笑话主题（用户输入）
feedback: str # 评估给出的改进意见
funny_or_not: str # 评估结果：funny / not funny
# 节点定义
def llm_call_generator(state: State):
"""生成器节点：生成笑话，如果有历史反馈就参考反馈重新生成"""
# 如果存在反馈，就结合反馈优化笑话；否则直接写新笑话
if state.get("feedback"):
msg = llm.invoke(
f"写一个关于{state['topic']}的笑话，需要参考下面的反馈进行优化：{state['feedback']}"
)
else:
msg = llm.invoke(f"写一个关于{state['topic']}的笑话")
return {"joke": msg.content}
def llm_call_evaluator(state: State):
"""评估器节点：对笑话打分，输出结构化评价结果"""
grade = evaluator.invoke(f"评价这个笑话：{state['joke']}")
return {"funny_or_not": grade.grade, "feedback": grade.feedback}
def route_joke(state: State):
"""条件路由函数：根据评估结果决定流程走向
Accepted：笑话合格，结束任务
Rejected + Feedback：笑话不合格，返回生成器重写
"""
if state["funny_or_not"] == "funny":
return "Accepted"
elif state["funny_or_not"] == "not funny":
return "Rejected + Feedback"
if __name__ == "__main__":
# 构建工作流
optimizer_builder = StateGraph(State)
# 添加节点
optimizer_builder.add_node("llm_call_generator", llm_call_generator)
optimizer_builder.add_node("llm_call_evaluator", llm_call_evaluator)
# 边配置
optimizer_builder.add_edge(START, "llm_call_generator")
optimizer_builder.add_edge("llm_call_generator", "llm_call_evaluator")
# 条件边：评估完成后动态分支
optimizer_builder.add_conditional_edges(
"llm_call_evaluator",
route_joke,
{
"Accepted": END, # 笑话合格 → 结束
"Rejected + Feedback": "llm_call_generator" # 不合格 → 回到生成器重新写笑话
},
)
# 编译图
optimizer_workflow = optimizer_builder.compile()
# 启动循环优化工作流
result = optimizer_workflow.invoke({"topic": "程序员"})
print("最终结果")
print(f"主题：{result['topic']}")
print(f"最终笑话：{result['joke']}")
print(f"评估结果：{result['funny_or_not']}")
print(f"评估反馈：{result['feedback']}")
実行するコードは次の情報を出力します。
最终结果
主题：程序员
最终笑话：程序员的笑话：
程序员的笑话是这样的：
评估结果：funny
评估反馈：程序员的笑话确实很有趣，因为它重复了“程序员的笑话是这样的”这句话，形成了一个循环。这种重复和循环的结构使得笑话更加生动和有趣。
以前紹介したチェーン、ブランチ、ループ、レイアウト、評価オプティマイザーのワークフローは、ほとんどの流れが事前に定義されています。知的体（エージェント）と呼ばれるツールの方が独立した意思決定能力が高いのに対し、大型モデルはそのツールを呼び出すかどうか、どのツールを呼び出すか、何度も呼び出すかなどを独立して決定します。問題と解決策を事前に予測できないシナリオに適しています。典型的なのは、ReAct (Reasoning + Action) 范式、外部アクションの実行中に、外部アクションを実行しながら（駐在意设、滞在意発信、外部アクションを実行しながら（駐在意発行）滞在意论、駐在设计」ループを、知覚意思側のモデルとして実行します。
全体的な構造のフローチャート:
参照コード構造:
from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage, ToolMessage
from langchain_core.tools import tool
# ---------------------------------------------------
# 定义工具
# ---------------------------------------------------
@tool
def get_current_time() -> str:
"""获取当前系统时间，当用户询问现在几点、当前时间的时候使用该工具"""
from datetime import datetime
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
@tool
def calculator(a: float, b: float, op: str) -> str:
"""
简单计算器，做加减乘除运算
Args:
a: 第一个数字
b: 第二个数字
op: 运算符，可选 + - * /
"""
if op == "+":
res = a + b
elif op == "-":
res = a - b
elif op == "*":
res = a * b
elif op == "/":
res = a / b
else:
return "错误：不支持的运算符"
return f"计算结果：{a} {op} {b} = {res}"
# 指定工具列表，包含所有定义的工具
tools = [get_current_time, calculator]
# 绑定工具，决定是否要调用工具
def agent_node(state: MessagesState):
llm = ChatOpenAI(
model="qwen2.5-1.5b-instruct-q4_k_m.gguf",
base_url="
api_key="dummy",
temperature=0.3,
max_tokens=512,
)
# 将工具集合绑定给LLM大模型
llm_with_tools = llm.bind_tools(tools)
sys_msg = SystemMessage(content="""
你是助手小张。可以使用工具完成任务。
- 问时间，调用 get_current_time
- 数学计算，调用 calculator
- 普通聊天，直接回答，不要强行调用工具
""")
# 消息列表，包含系统消息和用户消息
messages = [sys_msg] + state["messages"]
response = llm_with_tools.invoke(messages)
return {"messages": [response]}
# 工具执行节点：执行LLM生成的tool_calls
def tool_executor_node(state: MessagesState):
# 把工具放到字典方便查找
tool_map = {t.name: t for t in tools}
last_ai_msg = state["messages"][-1]
tool_messages = []
for tool_call in last_ai_msg.tool_calls:
tool_name = tool_call["name"]
tool_args = tool_call["args"]
selected_tool = tool_map[tool_name]
# 执行工具
tool_result = selected_tool.invoke(tool_args)
tool_messages.append(
ToolMessage(content=str(tool_result), tool_call_id=tool_call["id"])
)
return {"messages": tool_messages}
# 条件路由：判断是否需要走工具路径
def should_use_tools(state: MessagesState):
last_msg = state["messages"][-1]
# 如果AI消息包含tool_calls，去执行工具；否则结束
if last_msg.tool_calls:
return "use_tool"
else:
return "finish"
if __name__ == "__main__":
# 构建工作流
builder = StateGraph(MessagesState)
# 添加节点
builder.add_node("agent", agent_node)
builder.add_node("tool_executor", tool_executor_node)
# 起始边
builder.add_edge(START, "agent")
# 条件分支：agent出来之后判断是否调用工具
builder.add_conditional_edges(
source="agent",
path=should_use_tools,
path_map={
"use_tool": "tool_executor",
"finish": END
}
)
# 工具执行完，回到agent，让LLM基于工具结果生成最终回答
builder.add_edge("tool_executor", "agent")
graph = builder.compile()
# 交互式控制台多轮对话
current_state = {"messages": []}
while True:
user_input = input("\n你：")
if user_input.strip().lower() == "quit":
print("退出程序")
break
current_state = graph.invoke({
**current_state,
"messages": current_state["messages"] + [HumanMessage(content=user_input)]
})
ai_reply = current_state["messages"][-1]
print(f"小张：{ai_reply.content}")
出力を実行するコードには、次のメッセージが表示されます。
对话已启动，输入 quit 退出
你：计算1+1
小张：计算结果是：1.0 + 1.0 = 2.0
你：quit
退出程序
