LangGraphで自律制御のストーリーメーカーを作る
AIエージェントというのが注目されてきています。AIエージェントは与えられた目的に対して自律的に判断し次々にタスクを繰り返すことでより精度の高いアウトプットを行うことができます。
チャット入力欄に「ああでもない、こうでもない」とダメ出しを続けて質を上げていくのに疲れてしまった方にこそおすすめなのが、このAIエージェントです。
LangGraphを使うと、いい感じに自律的にタスクを繰り返してくれる機能を実装することができます。
作るもの：5つのグラフ
ざっくりとした小説のアイデアを受け取り、簡単にストーリーの起承転結の構造を作ってくれるグラフ（generate_structure）
起承転結の構造に、小説を面白くするアイデアを追加してくれるグラフ（analyze_and_add_gimmicks）
できた小説を評価し、改善のアイデアがあれば３つ挙げ、十分なら次のグラフに渡してくれるグラフ（evaluate_idea）
３つの改善案を取り入れて小説を改善してくれるグラフ（revise_story）
改善された小説を改めて起承転結に書き起こし完成させてくれるグラフ（finalize_story）
使い方
次のように使います
python story_maker.py "桃太郎が実は鬼の子であったというラノベ。昔話桃太郎の後日譚" --max_api_calls 7pythonコード実行時に、ざっくりどんな物語かを引数で渡します
APIをコールする上限数を指定します
得られる結果
2024-07-17 19:51:03,662 - INFO - API Call Count (generate_structure): 1
2024-07-17 19:51:11,240 - INFO - HTTP Request: POST https://api.openai.com/v1/chat/completions "HTTP/1.1 200 OK"
2024-07-17 19:51:11,251 - INFO - Structure generation completed
2024-07-17 19:51:11,255 - INFO - API Call Count (analyze_and_add_gimmicks): 2
2024-07-17 19:51:17,096 - INFO - HTTP Request: POST https://api.openai.com/v1/chat/completions "HTTP/1.1 200 OK"
2024-07-17 19:51:17,102 - INFO - Gimmick analysis completed
2024-07-17 19:51:17,108 - INFO - API Call Count (evaluate_idea): 3
2024-07-17 19:51:28,155 - INFO - HTTP Request: POST https://api.openai.com/v1/chat/completions "HTTP/1.1 200 OK"
2024-07-17 19:51:28,163 - INFO - API Call Count (finalize_story): 4
2024-07-17 19:52:02,972 - INFO - HTTP Request: POST https://api.openai.com/v1/chat/completions "HTTP/1.1 200 OK"
2024-07-17 19:52:02,975 - INFO - Evaluation and revision completed
Novel Generation Process:
Final Story:
起：昔々、日本のある村に桃太郎という少年がいた。彼は村人たちから勇敢で正義感が強いと称えられ、鬼退治を行うことで有名となった。しかし、ある日、村の長老から衝撃的な秘密を告げられる。それは、桃太郎が実は鬼の子であるという事実だった。
承：桃太郎はその衝撃的な事実を受け入れるのに時間がかかった。彼は村との関係を保つため、自分が鬼の子であることを隠し、普通の少年として生活を続けた。しかし、鬼としての力が次第に覚醒し始め、それを制御するのに苦労した。自分の正体を隠すことの窮屈さと、自分の力を制御するための訓練に励む日々が続いた。
転：ついに桃太郎は村人たちに自分が鬼の子であることを打ち明ける決意をする。村人たちは最初は驚きと不信感で彼を見つめたが、桃太郎の誠実さと勇気に触れ、徐々に受け入れ始める。桃太郎は自分の力をコントロールする訓練を重ね、鬼としての力を使いこなせるようになる。そして、鬼と人間の共存を目指す旅に出る覚悟を固める。
結：桃太郎は鬼としての力を使い、鬼と人間の共存を訴える旅に出る。その途中で新しい仲間と出会い、彼らと共に多くの困難を乗り越える。そして、人間と鬼が互いを理解し合う未来を作るために、自身のアイデンティティを見つけていく。そうして、桃太郎は自分の運命を受け入れ、鬼と人間の共存を目指す新たな伝説を築き上げた。彼の物語は、勇敢さと正義感、そして自分自身を受け入れる強さを持つことの大切さを教えてくれる。
Total API Calls: 4
Story Versions:
Version 1:
起：昔々、桃太郎は日本の村で生まれた。彼は勇敢で正義感が強く、鬼退治を行うことで有名だった。しかし、ある日、村の長老から驚くべき秘密を告げられる。なんと、桃太郎は実は鬼の子であるというのだ。
承：桃太郎はその事実を受け入れるのに時間がかかった。彼は自分が鬼の子であることを隠し、村の人々との関係を守るために奮闘した。しかし、次第に彼の中で葛藤が生まれ始める。彼は自分が本当の正体を隠すことに疲れ果て、決意を固める。
転：桃太郎は鬼の血を引いていることを公にし、村の人々に受け入れてもらおうとする。最初は驚きと不信感を持つ者もいたが、桃太郎の誠実さと勇気に触れ、徐々に彼を受け入れ始める。桃太郎は自分の運命と向き合い、新たな旅に出る覚悟を固める。
結：桃太郎は鬼の子としての力を使い、鬼と人間の共存を目指す旅に出る。彼は新たな仲間と出会い、多くの試練に立ち向かいながら、鬼と人間が互いを理解し合う未来を切り開いていく。桃太郎は自分の運命を受け入れ、新たな伝説を築いていくのだった。
Version 2:
アイデア：
1. 桃太郎が鬼の力を使うことで、新たな戦い方や能力を身につける過程を描く。彼がその力をコントロールするために訓練を受けるシーンを追加することで、物語にスリリングな要素を加える。
2. 鬼と人間の間で生じる葛藤や対立をより複雑に描写し、両者が互いを理解し合う過程で起こるドラマを強調する。特に、桃太郎がどちらの側に立つべきか迷いながら、自分のアイデンティティを見つける過程を詳細に描写する。
3. 新たな仲間との出会いを通じて、桃太郎が成長していく姿を描く。仲間たちとの絆が彼の旅において重要な役割を果たすことで、物語に温かさや感動を加える。
4. 結末で、桃太郎が鬼と人間の共存を実現するために奮闘する姿を描くだけでなく、彼が自分の運命を受け入れる過程や成長を強調する。彼が新たな伝説を築くことで、読者に希望や勇気を与えるような結末にすることで、物語の感動を高める。
Version 3:
起：昔々、日本のある村に桃太郎という少年がいた。彼は村人たちから勇敢で正義感が強いと称えられ、鬼退治を行うことで有名となった。しかし、ある日、村の長老から衝撃的な秘密を告げられる。それは、桃太郎が実は鬼の子であるという事実だった。
承：桃太郎はその衝撃的な事実を受け入れるのに時間がかかった。彼は村との関係を保つため、自分が鬼の子であることを隠し、普通の少年として生活を続けた。しかし、鬼としての力が次第に覚醒し始め、それを制御するのに苦労した。自分の正体を隠すことの窮屈さと、自分の力を制御するための訓練に励む日々が続いた。
転：ついに桃太郎は村人たちに自分が鬼の子であることを打ち明ける決意をする。村人たちは最初は驚きと不信感で彼を見つめたが、桃太郎の誠実さと勇気に触れ、徐々に受け入れ始める。桃太郎は自分の力をコントロールする訓練を重ね、鬼としての力を使いこなせるようになる。そして、鬼と人間の共存を目指す旅に出る覚悟を固める。
結：桃太郎は鬼としての力を使い、鬼と人間の共存を訴える旅に出る。その途中で新しい仲間と出会い、彼らと共に多くの困難を乗り越える。そして、人間と鬼が互いを理解し合う未来を作るために、自身のアイデンティティを見つけていく。そうして、桃太郎は自分の運命を受け入れ、鬼と人間の共存を目指す新たな伝説を築き上げた。彼の物語は、勇敢さと正義感、そして自分自身を受け入れる強さを持つことの大切さを教えてくれる。Final Story
AIエージェントによって改善された完成原稿です
Total API Calls
OpenAIのAPIをコールした回数です
Story Versions
初期稿から改善されていく様子を確認できます
コード実行の準備（ライブラリのインストール）
（必要に応じてお使いください）
$ pip install -r requirements.txtrequirements.txt
annotated-types==0.7.0
anyio==4.4.0
asttokens==2.4.1
certifi==2024.7.4
charset-normalizer==3.3.2
decorator==5.1.1
distro==1.9.0
executing==2.0.1
graphviz==0.20.3
h11==0.14.0
httpcore==1.0.5
httpx==0.27.0
idna==3.7
ipython==8.26.0
jedi==0.19.1
jsonpatch==1.33
jsonpointer==3.0.0
langchain-core==0.2.19
langchain-openai==0.1.16
langgraph==0.1.8
langsmith==0.1.86
matplotlib-inline==0.1.7
openai==1.35.14
orjson==3.10.6
packaging==24.1
parso==0.8.4
pexpect==4.9.0
prompt_toolkit==3.0.47
ptyprocess==0.7.0
pure-eval==0.2.2
pydantic==2.8.2
pydantic_core==2.20.1
Pygments==2.18.0
PyYAML==6.0.1
regex==2024.5.15
requests==2.32.3
six==1.16.0
sniffio==1.3.1
stack-data==0.6.3
tenacity==8.5.0
tiktoken==0.7.0
tqdm==4.66.4
traitlets==5.14.3
typing_extensions==4.12.2
urllib3==2.2.2
wcwidth==0.2.13dotenvの設定
story_maker.pyと同じディレクトリに、.env ファイルを作成し、openaiのAPIキーを記入しておきます
OPENAI_API_KEY=sk-XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXstory_maker.py
import os
import logging
import argparse
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langchain_core.messages import HumanMessage, AIMessage
from typing import Dict, Any, List
from typing_extensions import TypedDict
# 環境変数の読み込み
load_dotenv()
# ロギングの設定
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
# 状態を保持するための型定義
class State(TypedDict):
messages: List[HumanMessage]
api_call_count: int
story_versions: List[str]
# GPTインスタンスの初期化
llm_gpt3 = ChatOpenAI(temperature=0.7, model="gpt-3.5-turbo")
llm_gpt4 = ChatOpenAI(temperature=0.7, model="gpt-4")
# API呼び出し回数を管理するデコレータ関数
def api_call_wrapper(func):
def wrapper(state: State):
state["api_call_count"] = state.get("api_call_count", 0) + 1
logging.info(f"API Call Count ({func.__name__}): {state['api_call_count']}")
return func(state)
return wrapper
# 小説の構造を生成する関数
@api_call_wrapper
def generate_structure(state: State) -> Dict[str, Any]:
structure_prompt = HumanMessage(content=f"このプロンプトに基づいて小説の起承転結を執筆してください: {state['messages'][0].content}")
response = llm_gpt3.invoke([structure_prompt])
state["story_versions"] = state.get("story_versions", []) + [response.content]
return {"messages": [response], "api_call_count": state["api_call_count"], "story_versions": state["story_versions"]}
# 小説にギミックを追加する関数
@api_call_wrapper
def analyze_and_add_gimmicks(state: State) -> Dict[str, Any]:
enhanced_message = HumanMessage(content=f"この小説の構造を分析し、面白くするためのアイデアを加筆してください: {state['messages'][0].content}")
response = llm_gpt3.invoke([enhanced_message])
state["story_versions"] = state.get("story_versions", []) + [response.content]
return {"messages": [response], "api_call_count": state["api_call_count"], "story_versions": state["story_versions"]}
# 小説のアイデアを評価する関数
@api_call_wrapper
def evaluate_idea(state: State) -> Dict[str, Any]:
evaluation_message = HumanMessage(content=f"以下の内容を精査してから、評価してください。読者を驚かせたり、気になって続きを読みたくなったりさせるための、改善のアイデアがあれば３つ挙げてください。もう十分に面白ければ完成と言ってください: {state['messages'][0].content}")
response = llm_gpt3.invoke([evaluation_message])
return {"messages": [response], "api_call_count": state["api_call_count"], "story_versions": state["story_versions"]}
# 小説を改訂する関数
@api_call_wrapper
def revise_story(state: State) -> Dict[str, Any]:
revision_message = HumanMessage(content=f"次の改善点に基づいてこの小説を修正してください。: {state['messages'][0].content}")
response = llm_gpt3.invoke([revision_message])
state["story_versions"] = state.get("story_versions", []) + [response.content]
return {"messages": [response], "api_call_count": state["api_call_count"], "story_versions": state["story_versions"]}
# 完成版小説を執筆する関数
@api_call_wrapper
def finalize_story(state: State) -> Dict[str, Any]:
final_story_prompt = HumanMessage(content=f"以下の改善案を反映して、小説を起承転結にまとめてください。小説を面白く読ませるに十分な情報量にしてください。: {' '.join(state['story_versions'])}")
response = llm_gpt4.invoke([final_story_prompt])
state["story_versions"] = state.get("story_versions", []) + [response.content]
return {"messages": [response], "api_call_count": state["api_call_count"], "story_versions": state["story_versions"]}
# グラフを作成する関数
def create_graph(node_func, name):
graph = StateGraph(State)
graph.add_node(name, node_func)
graph.set_entry_point(name)
graph.set_finish_point(name)
return graph.compile()
# 構造生成とギミック追加のランナーを作成
structure_runner = create_graph(generate_structure, "generate_structure")
gimmicks_runner = create_graph(analyze_and_add_gimmicks, "analyze_and_add_gimmicks")
# 評価と改訂のためのグラフを作成
evaluation_graph = StateGraph(State)
evaluation_graph.add_node("evaluate_idea", evaluate_idea)
evaluation_graph.add_node("revise_story", revise_story)
evaluation_graph.add_node("finalize_story", finalize_story)
# 評価ルーター関数を定義
def evaluation_router(state: State, config: Dict[str, Any]):
max_api_calls = config.get("max_api_calls", 7) # デフォルト値を7に設定
if state["api_call_count"] >= max_api_calls or "完成" in state["messages"][-1].content:
return "finalize_story"
else:
return "revise_story"
# グラフのエントリーポイントとエッジを設定
evaluation_graph.set_entry_point("evaluate_idea")
evaluation_graph.add_conditional_edges(
"evaluate_idea",
evaluation_router,
{"revise_story": "revise_story", "finalize_story": "finalize_story"}
)
evaluation_graph.add_edge("revise_story", "evaluate_idea")
evaluation_graph.add_edge("finalize_story", END)
# 評価ランナーをコンパイル
evaluation_runner = evaluation_graph.compile()
# メッセージをフォーマットする関数
def format_message(message: Dict[str, Any]) -> str:
return f"Human: {message.content}" if isinstance(message, HumanMessage) else f"AI: {message.content}"
# 出力をフォーマットする関数
def format_output(response: Dict[str, Any]) -> str:
output = "Novel Generation Process:\n\n"
output += f"Final Story:\n{response['messages'][-1].content}\n\n"
output += f"Total API Calls: {response['api_call_count']}\n\n"
output += "Story Versions:\n"
for i, version in enumerate(response['story_versions'], 1):
output += f"\nVersion {i}:\n{version}\n"
return output
# 小説生成プロセスを実行する関数
def run_novel_generation(prompt: str, max_api_calls: int):
# 初期入力を設定
initial_input = {"messages": [HumanMessage(content=prompt)], "api_call_count": 0, "story_versions": []}
# 構造生成を実行
structure_response = structure_runner.invoke(initial_input)
logging.info("Structure generation completed")
# ギミック追加を実行
gimmicks_response = gimmicks_runner.invoke(structure_response)
logging.info("Gimmick analysis completed")
# 評価と改訂を実行
try:
evaluation_response = evaluation_runner.invoke(gimmicks_response, {"max_api_calls": max_api_calls})
logging.info("Evaluation and revision completed")
return format_output(evaluation_response)
except Exception as e:
logging.error(f"Error during evaluation and revision: {str(e)}")
return format_output(gimmicks_response)
# メイン関数
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Novel Idea Generator")
parser.add_argument("prompt", type=str, help="The prompt for the novel idea")
parser.add_argument("--max_api_calls", type=int, default=7, help="Maximum number of API calls")
args = parser.parse_args()
result = run_novel_generation(args.prompt, args.max_api_calls)
print(result)あとはお好きに使ってみてください
python story_maker.py "浦島太郎が泳げなくて竜宮城に辿り着けない話" --max_api_calls 7