5.1 设计思路

5.2 实现方案

import json
from ENV import deep_seek_url, deep_seek_api_key, deep_seek_default_model
import Agently
import os 

# 创建一个作家agent

writer = (
    Agently.create_agent()
        .set_settings("current_model", "OAIClient")
        .set_settings("model.OAIClient.url", os.environ["DEEPSEEK_BASE_URL"])
        .set_settings("model.OAIClient.auth", { "api_key": os.environ["DEEPSEEK_API_KEY"] })
        .set_settings("model.OAIClient.options", { "model": os.environ["DEEP_SEEK_DEFAULT_MODEL"] })
)

# 创建两个工作流：主工作流和分块创作工作流
main_workflow = Agently.Workflow()
block_workflow = Agently.Workflow()

# 定义主工作流的工作块
## 输入一句话描述
@main_workflow.chunk()
def input_story_idea(inputs, storage):
    storage.set("story_idea", input("[💡请输入您的故事灵感]: "))
    return

## 创建世界观背景故事
@main_workflow.chunk()
def generate_background(inputs, storage):
    story_idea = storage.get("story_idea")
    background = (
        writer
            .input({
                "故事灵感": story_idea
            })
            .instruct(
"""请根据{故事灵感}创作故事的世界信息和背景故事，其中：
世界信息需要包括世界的主要国家或地区分布，不同国家或地区的环境描写，科技水平，信仰情况等
世界背景故事需要以时间线的形式描述世界的主要历史沿革，国家或地区之间的重大事件及带来的影响变化等"""
            )
            .output({
                "世界名称": ("str", ),
                "主要国家或地区": [{
                    "名称": ("str", ),
                    "关键信息": ("str", ),
                }],
                "世界背景故事": [("str", )],
            })
            .start()
    )
    storage.set("background", background)
    return {
        "title": "世界观背景故事",
        "result": background,
    }

## 创建关键情节线
@main_workflow.chunk()
def generate_storyline(inputs, storage):
    story_idea = storage.get("story_idea")
    background = storage.get("background")
    storyline = (
        writer
            .input({
                "故事灵感": story_idea,
                "世界观背景故事": background,
            })
            .instruct(
"""请根据{世界观背景故事}，围绕{故事灵感}，创作故事的关键情节线安排"""
            )
            .output({
                "情节结构类型": ("str", "基于常见的故事、小说、剧作创作方法，输出你将要使用的剧情结构类型名称"),
                "情节结构特点": ("str", "阐述{剧情结构类型}的剧情结构手法、特点"),
                "故事线详细创作": [{
                    "本段故事作用": ("str", "描述本段故事在整体结构中发挥的作用"),
                    "关键情节": ([("str", )], "按时序描述本段故事中的关键情节，以及情节中的关键细节"),
                    "涉及关键人物": ([("str", )], "给出本段故事中涉及的关键人物名"),
                }],
            })
            .start()
    )
    storage.set("storyline", storyline)
    return {
        "title": "关键情节线",
        "result": storyline,
    }

## 分发故事段落设计
@main_workflow.chunk()
def send_story_block_list(inputs, storage):
    storyline = storage.get("storyline")
    storyline_details = storyline["故事线详细创作"]
    extra_instruction = input("[您是否还有其他创作指导说明？如创作风格、注意事项等]")
    story_block_list = []
    for item in storyline_details:
        item.update({ "补充创作指导": extra_instruction })
        story_block_list.append(item)
    return story_block_list

## 过程产出输出
@main_workflow.chunk_class()
def print_process_output(inputs, storage):
    print(f"[{ inputs['default']['title'] }]:")
    if isinstance(inputs["default"]["result"], dict):
        print(
            json.dumps(inputs["default"]["result"], indent=4, ensure_ascii=False)
        )
    else:
        print(inputs["default"]["result"])
    return

## 最终结果整理
@main_workflow.chunk()
def sort_out(inputs, storage):
    result = []
    for item in inputs["default"]:
        result.append(item["default"])
    return "\n\n".join(result)

# 定义分块创作工作流的工作块
## 获取初始数据
@block_workflow.chunk()
def init_data(inputs, storage):
    storage.set("story_block", inputs["default"])
    # 从公共存储中取出上一段创作结果
    storage.set("last_block_content", block_workflow.public_storage.get("last_block_content"))
    return

## 进行正文创作
@block_workflow.chunk()
def generate_block_content(inputs, storage):
    # 要考虑的条件较多，可以在请求外部构造input和instruct的prompt数据
    ## 围绕故事线详细创作信息的prompt
    ## {"本段故事作用": ,"关键情节": , "涉及关键人物": , "补充创作指导": }
    input_dict = storage.get("story_block")
    instruction_list = [
        "参考{本段故事作用}及{涉及关键人物}，将{关键情节}扩写为完整的故事",
        "每段故事需要尽量包括行动描写、心理活动描写和对白等细节",
        "每次创作只是完整文章结构中的一部分，承担{本段故事作用}说明的作用任务，只需要按要求完成{关键情节}的描述即可，不需要考虑本段故事自身结构的完整性",
    ]
    
    ## 如果有前一段内容，通过传入前一段内容末尾确保创作的连贯性
    last_block_content = storage.get("last_block_content", None)
    if last_block_content:
        ## 在这里取上一段落的最后50个字，可根据需要修改保留的长度
        keep_length = 50
        input_dict.update({ "上一段落的末尾": last_block_content[(-1 * keep_length):] })
        instruction_list.append("创作时需要承接{上一段落的末尾}，确保表达的连贯性")
    
    ## 如果有人类评判反馈的修改意见，添加prompt
    last_generation = storage.get("block_content", None)
    revision_suggestion = storage.get("revision_suggestion", None)
    if last_generation and revision_suggestion:
        input_dict.update({
            "已有创作结果": last_generation,
            "修改意见": revision_suggestion,
        })
        instruction_list.append("你之前已经创作了{已有创作结果}，但仍然需要修改，请参考{修改意见}进行修订")
    
    # 开始创作
    block_content = (
        writer
            .input(input_dict)
            .instruct(instruction_list)
            .start()
    )

    # 保存创作结果
    storage.set("block_content", block_content)
    return {
        "title": f"本轮创作目标：{ input_dict['本段故事作用'] }",
        "result": block_content,
    }

## 人类判断是否满意
@block_workflow.chunk()
def human_confirm(inputs, storage):
    confirm = ""
    while confirm.lower() not in ("y", "n"):
        confirm = input("[您是否满意本次创作结果？(y/n)]: ")
    return confirm.lower()

## 提交修改意见
@block_workflow.chunk()
def input_revision_suggestion(inputs, storage):
    storage.set("revision_suggestion", input("[请输入您的修改意见]: "))
    return

## 输出满意的创作成果
@block_workflow.chunk()
def return_block_content(inputs, storage):
    block_content = storage.get("block_content")
    # 记得在公共存储中更新本次创作结果
    block_workflow.public_storage.set("last_block_content", block_content)
    return block_content

## 过程产出输出
@block_workflow.chunk_class()
def print_process_output(inputs, storage):
    print(f"[{ inputs['default']['title'] }]:")
    if isinstance(inputs["default"]["result"], dict):
        print(
            json.dumps(inputs["default"]["result"], indent=4, ensure_ascii=False)
        )
    else:
        print(inputs["default"]["result"])
    return

# 定义分块创作工作流的工作流程
(
    block_workflow
        .connect_to("init_data")
        .connect_to("generate_block_content")
        .connect_to("@print_process_output")
        .connect_to("human_confirm")
        .if_condition(lambda return_value, storage: return_value == "y")
            .connect_to("return_block_content")
            .connect_to("end")
        .else_condition()
            .connect_to("input_revision_suggestion")
            .connect_to("generate_block_content")
)

(
    main_workflow
        .connect_to("input_story_idea")
        .connect_to("generate_background")
        .connect_to("@print_process_output")
        .connect_to("generate_storyline")
        .connect_to("@print_process_output")
        .connect_to("send_story_block_list")# -> list[item1, item2, item3, ...]
            .loop_with(block_workflow) # item1 -> block_workflow:inputs["default"]; item2 -> block_workflow: i
        #.connect_to("sort_out")
        .connect_to("end")
)

# 打印流程图，检查流程正确性
print(main_workflow.draw())

result = main_workflow.start()
print(result["default"])

5.3 运行示例

以"小老虎的花园"为故事灵感，运行工作流的完整输出：

[💡请输入您的故事灵感]: 小老虎的花园
[世界观背景故事]:
{
    "世界名称": "小老虎的花园",
    "主要国家或地区": [
        {
            "名称": "花香国",
            "关键信息": "花香国是一个以花卉为主要产业的国家，处处皆是花园，人们热爱花草，工艺精湛，以花艺闻名世界。科技水平较发达，但环保意识深厚。信仰主要是自然神灵和花神。"
        },
        {
            "名称": "绿野地",
            "关键信息": "绿野地是一个葱郁的绿色国家，大片原始森林和草原遍布全境，生态优美。科技相对落后，但人民纯朴善良，信仰多元，崇尚自然。"
        }
    ],
    "世界背景故事": [
        "500年前，花香国和绿野地之间曾经有过一场激烈的领土争夺战，最终导致花香国夺得了一块富饶的土地，而且绿野地的领土遭受了严重的破坏。",
        "100年前，花香国的一位名叫花语的花艺大师研发出了一种能够让花朵永不凋谢的魔法药剂，从此让花香国的花卉产业大放异彩，成为世界花卉的主要供应国。",
        "30年前，花香国和绿野地之间爆发了一场环保危机，花香国的工业发展导致大量污染物排放，绿野地的植被开始受到影响。最终在国际社会的压力下，花香国进行了环保整治，两国之间的关系也因此得以改善。"
    ]
}
[关键情节线]:
{
    "情节结构类型": "传统冲突决策结构",
    "情节结构特点": "主要通过角色之间的冲突和抉择来推动故事情节发展，最终达到解决矛盾的目的。",
    "故事线详细创作": [
        {
            "本段故事作用": "介绍花语和小老虎之间的深厚友谊",
            "关键情节": [
                "花语是花香国的著名花艺大师，她在小老虎的花园里发现了一种奇特的花朵，对其产生浓厚兴趣。",
                "小老虎是花香国的一位年轻花农，专心经营着自己的花园，对花卉有着独特的感悟。",
                "花语和小老虎在花园中相遇，彼此展示自己的花艺技巧，建立起深厚的友谊。",
                "他们共同研究出一种新的花卉种植方法，使得花园中的花朵更加美丽绚丽。"
            ],
            "涉及关键人物": [
                "花语",
                "小老虎"
            ]
        },
        {
            "本段故事作用": "揭示花香国和绿野地之间的历史纠葛",
            "关键情节": [
                "花语的研发的永不凋谢的魔法药剂被偷窃，引发了花香国和绿野地之间的一场纷争。",
                "花香国和绿野地的代表在边界线上举行谈判，试图解决魔法药剂被盗事件。",
                "谈判过程中，双方回顾了500年前的领土争夺战，导致两国之间的关系再度紧张。",
                "最终，通过调解和妥协，双方达成协议，开始合作解决魔法药剂遗失的问题。"
            ],
            "涉及关键人物": [
                "花语",
                "花香国代表",
                "绿野地代表"
            ]
        },
        {
            "本段故事作用": "探讨环保与发展的关系",
            "关键情节": [
                "花语和小老虎发现花园中的花朵开始凋零，怀疑是环境问题造成的。",
                "他们一起前往花香国进行调查，发现工业污染导致了周围环境的恶化。",
                "花语和小老虎决定与当地政府合作，推动环保运动，改善环境质量。",
                "通过大家的努力，花香国的环保意识得到提升，与绿野地之间的合作也更加密切。"
            ],
            "涉及关键人物": [
                "花语",
                "小老虎",
                "当地政府代表"
            ]
        }
    ]
}
[您是否还有其他创作指导说明？如创作风格、注意事项等]
[本轮创作目标：介绍花语和小老虎之间的深厚友谊]:
花语是花香国的著名花艺大师，她以独特的花艺技巧闻名于世。一天，她听说了花香国某处花园中有一种奇特的花朵，对其产生了浓厚的兴趣。于是，她决定前往探寻这种神秘的花朵。

而花园的主人正是小老虎，一个勤劳且热爱花卉的年轻花农。他专心经营着自己的花园，对花卉有着独特的感悟，喜欢在花园里静静地品味花朵的香气。

当花语来到小老虎的花园时，被眼前绚丽多彩的花朵所震撼。小老虎也被花语所展现的花艺技巧深深吸引。两人开始互相交流，彼此分享各自对花卉的热爱和心得体会，渐渐地建立起了深厚的友谊。

他们在花园里共同研究出一种新的花卉种植方法，结合了花语的花艺技巧和小老虎的种植经验，使得花园中的花朵在色彩和芬芳上都更加出色。他们的合作不仅让花园变得更加美丽绚丽，也让彼此的友情日益加深。这对花语和小老虎来说，是一段珍贵的合作经历，也是一段难忘的友谊之旅。
[您是否满意本次创作结果？(y/n)]: y
[本轮创作目标：揭示花香国和绿野地之间的历史纠葛]:
花语和小老虎带着失落的心情回到花香国。他们发现花香国的人们正气愤地议论着魔法药剂被盗的事情。花语立即向花香国代表报告了情况，引起了一场紧急会议。会议上，花香国代表决定邀请绿野地的代表来进行谈判，希望通过友好的方式解决这起纠纷。

当绿野地代表抵达边界线时，花香国代表和他进行了深入的交谈。双方回顾了500年前的领土争夺战，让彼此的情绪再度升温。花香国代表强调，花语的研发是花香国的骄傲，不能容忍有人侵犯。而绿野地代表则表示，绿野地对这起事件一无所知，愿意与花香国合作找出真相。

经过长时间的谈判和协商，双方最终达成了协议。他们决定共同展开调查，找出魔法药剂的下落。这次合作不仅解决了魔法药剂被盗的问题，也为花香国和绿野地之间的关系奠定了基础。花语和小老虎感到欣慰，这次经历不仅让他们成长，也加深了彼此之间的友谊。整个花香国也因为这个事件而显得更加团结和坚强。
[您是否满意本次创作结果？(y/n)]: y
[本轮创作目标：探讨环保与发展的关系]:
花语和小老虎决定动身前往花香国，他们在路上谨慎地计划着如何进行调查。当他们到达花香国时，他们被眼前的景象所震惊，一片本应繁花似锦的花园却充斥着凋零的花朵和鬼魅般的树木。

花语低声说道："看来这里的环境出了问题，我们得找出原因才行。"小老虎点了点头，两人决定先去当地政府了解情况。在政府大楼门口，他们花了不少时间才见到了当地政府代表。

经过一番交谈，他们得知这片土地受到了工业污染的严重影响，导致了植物凋零、空气污染等问题。花语和小老虎决定与政府代表合作，发起一场环保运动，希望改善这里的环境质量。

他们开始了环保宣传活动，号召当地居民一起加入到环保行动中来。逐渐地，越来越多的人参与到环保运动中，树立起环保意识。花语和小老虎感到欣慰，他们的努力正在取得成果，花香国的环境也逐渐得到改善。

同时，与绿野地之间的合作也更加密切，大家共同努力着让这片土地重新焕发生机。这次的经历不仅让花语和小老虎成长了许多，也加深了他们之间的友谊。整个花香国也因为这个事件变得更加团结和坚强。
[您是否满意本次创作结果？(y/n)]: y

工作流成功完成了三段故事的创作，每段都经过人类反馈确认后才进入下一段的创作。