AutoGen创意写作：文学创作智能体

【免费下载链接】autogen 启用下一代大型语言模型应用 项目地址: https://gitcode.com/GitHub_Trending/au/autogen

引言：当AI遇见文学创作

你是否曾经面对空白文档，灵感枯竭？或者想要创作一部小说却不知从何开始？AutoGen框架为文学创作带来了革命性的解决方案。通过多智能体协作，AutoGen能够模拟人类创作团队的工作流程，从构思到完稿，为作家、编剧和内容创作者提供强大的AI辅助创作能力。

本文将带你深入了解如何使用AutoGen构建专业的文学创作智能体系统，涵盖从基础设置到高级创作工作流的完整实现。

AutoGen文学创作系统架构

核心组件与功能模块

1. 创作智能体类型定义

智能体角色	职责描述	核心技术能力
创意总监	整体创作方向把控	主题理解、风格定位
情节设计师	故事结构设计	情节架构、冲突设置
人物塑造师	角色开发	性格塑造、关系网络
文风优化师	语言风格统一	文风模仿、一致性维护
章节创作师	具体内容生成	场景描写、对话创作
评审编辑	质量把控	逻辑检查、文笔优化

2. 基础环境配置

首先安装必要的依赖包：

# 安装AutoGen核心组件
pip install -U "autogen-agentchat" "autogen-ext[openai]"

# 可选：安装文本处理工具
pip install nltk spacy textstat
python -m spacy download en_core_web_sm

3. 文学创作智能体系统实现

3.1 基础创作智能体类

import asyncio
from typing import List, Dict, Any
from autogen_agentchat.agents import AssistantAgent
from autogen_ext.models.openai import OpenAIChatCompletionClient

class CreativeWritingAgent:
    def __init__(self, name: str, role: str, model: str = "gpt-4.1"):
        self.name = name
        self.role = role
        self.model_client = OpenAIChatCompletionClient(model=model)
        self.agent = self._create_agent()
    
    def _create_agent(self) -> AssistantAgent:
        system_message = f"""你是一名专业的{self.role}，擅长{self.get_role_expertise()}。
        请严格按照你的专业领域进行创作，保持风格一致性。"""
        
        return AssistantAgent(
            self.name,
            model_client=self.model_client,
            system_message=system_message,
            model_client_stream=True
        )
    
    def get_role_expertise(self) -> str:
        expertise_map = {
            "创意总监": "整体故事构思和主题定位",
            "情节设计师": "故事结构和情节发展设计",
            "人物塑造师": "角色性格和关系网络构建",
            "文风优化师": "语言风格统一和文笔优化",
            "章节创作师": "具体场景和对话内容创作",
            "评审编辑": "作品质量审核和修改建议"
        }
        return expertise_map.get(self.role, "文学创作")
    
    async def generate_content(self, prompt: str, context: Dict[str, Any] = None) -> str:
        full_prompt = self._build_prompt(prompt, context)
        response = await self.agent.run(task=full_prompt)
        return response
    
    def _build_prompt(self, prompt: str, context: Dict[str, Any]) -> str:
        if context:
            context_str = "\n".join([f"{k}: {v}" for k, v in context.items()])
            return f"创作上下文:\n{context_str}\n\n创作要求:\n{prompt}"
        return prompt
    
    async def close(self):
        await self.model_client.close()

3.2 多智能体协作创作系统

class LiteraryCreationSystem:
    def __init__(self):
        self.agents = {}
        self.workflow_stages = [
            "concept_development",
            "plot_design", 
            "character_creation",
            "writing_execution",
            "review_editing"
        ]
    
    async def initialize_agents(self):
        """初始化所有创作智能体"""
        agent_configs = [
            ("creative_director", "创意总监", "gpt-4.1"),
            ("plot_designer", "情节设计师", "gpt-4.1"),
            ("character_designer", "人物塑造师", "gpt-4.1"), 
            ("style_optimizer", "文风优化师", "gpt-4.1"),
            ("chapter_writer", "章节创作师", "gpt-4.1"),
            ("review_editor", "评审编辑", "gpt-4.1")
        ]
        
        for agent_id, role, model in agent_configs:
            self.agents[agent_id] = CreativeWritingAgent(agent_id, role, model)
    
    async def create_story(self, theme: str, genre: str, target_length: int = 50000) -> Dict[str, Any]:
        """完整的故事创作流程"""
        results = {}
        
        # 1. 概念开发阶段
        concept = await self._develop_concept(theme, genre, target_length)
        results['concept'] = concept
        
        # 2. 情节设计阶段
        plot = await self._design_plot(concept, genre)
        results['plot'] = plot
        
        # 3. 人物塑造阶段
        characters = await self._create_characters(concept, plot)
        results['characters'] = characters
        
        # 4. 章节创作阶段
        chapters = await self._write_chapters(plot, characters, target_length)
        results['chapters'] = chapters
        
        # 5. 评审编辑阶段
        final_work = await self._review_edit(chapters, concept)
        results['final_work'] = final_work
        
        return results
    
    async def _develop_concept(self, theme: str, genre: str, target_length: int) -> str:
        prompt = f"""基于主题"{theme}"和类型"{genre}"，创作一个完整的故事概念。
        目标字数：{target_length}字。
        包含：核心冲突、主题思想、目标读者群体。"""
        
        return await self.agents['creative_director'].generate_content(prompt)
    
    async def _design_plot(self, concept: str, genre: str) -> str:
        prompt = f"""根据以下故事概念设计完整的情节结构：
        {concept}
        
        类型：{genre}
        要求：三幕式结构，包含起承转合，明确的冲突和关键转折点。"""
        
        return await self.agents['plot_designer'].generate_content(prompt)
    
    async def _create_characters(self, concept: str, plot: str) -> str:
        prompt = f"""基于以下故事概念和情节设计主要角色：
        概念：{concept}
        情节：{plot}
        
        要求：主角、配角、反派角色完整档案，包含性格、背景、动机、成长弧光。"""
        
        return await self.agents['character_designer'].generate_content(prompt)
    
    async def _write_chapters(self, plot: str, characters: str, target_length: int) -> List[Dict]:
        chapters = []
        chapter_count = max(5, target_length // 5000)  # 每章约5000字
        
        for chapter_num in range(1, chapter_count + 1):
            prompt = f"""创作第{chapter_num}章内容：
            情节大纲：{plot}
            人物设定：{characters}
            
            要求：符合章节在整体结构中的位置，推进故事发展，展现人物性格。"""
            
            content = await self.agents['chapter_writer'].generate_content(prompt)
            chapters.append({
                'chapter_number': chapter_num,
                'content': content,
                'word_count': len(content.split())
            })
        
        return chapters
    
    async def _review_edit(self, chapters: List[Dict], concept: str) -> Dict[str, Any]:
        reviewed_chapters = []
        total_word_count = 0
        
        for chapter in chapters:
            prompt = f"""评审并优化以下章节内容：
            故事概念：{concept}
            章节内容：{chapter['content']}
            
            检查：逻辑一致性、文笔质量、节奏把控、错别字和语法错误。"""
            
            reviewed_content = await self.agents['review_editor'].generate_content(prompt)
            reviewed_chapters.append({
                'chapter_number': chapter['chapter_number'],
                'content': reviewed_content,
                'word_count': len(reviewed_content.split())
            })
            total_word_count += len(reviewed_content.split())
        
        return {
            'chapters': reviewed_chapters,
            'total_word_count': total_word_count,
            'chapter_count': len(reviewed_chapters)
        }
    
    async def close(self):
        """关闭所有智能体连接"""
        for agent in self.agents.values():
            await agent.close()

4. 高级创作功能扩展

4.1 风格模仿与迁移

class StyleTransferAgent:
    def __init__(self, base_agent: CreativeWritingAgent):
        self.base_agent = base_agent
    
    async def imitate_style(self, target_text: str, source_text: str, content: str) -> str:
        prompt = f"""模仿以下文本风格重写内容：
        目标风格文本：{target_text}
        源文本：{source_text}
        需要重写的内容：{content}
        
        要求：保持原意，但采用目标文本的语言风格、句式结构和修辞特点。"""
        
        return await self.base_agent.generate_content(prompt)
    
    async def analyze_writing_style(self, text: str) -> Dict[str, Any]:
        prompt = f"""分析以下文本的写作风格特征：
        {text}
        
        分析维度：句式长度、修辞手法、词汇特点、情感基调、节奏韵律。"""
        
        analysis = await self.base_agent.generate_content(prompt)
        return self._parse_style_analysis(analysis)
    
    def _parse_style_analysis(self, analysis: str) -> Dict[str, Any]:
        # 解析风格分析结果
        return {'raw_analysis': analysis}

4.2 情节冲突检测与优化

class PlotConflictDetector:
    def __init__(self, base_agent: CreativeWritingAgent):
        self.base_agent = base_agent
    
    async def detect_conflicts(self, plot: str, chapters: List[Dict]) -> List[Dict]:
        conflicts = []
        
        for i, chapter in enumerate(chapters):
            prompt = f"""检测以下情节与章节内容之间的冲突：
            情节大纲：{plot}
            第{i+1}章内容：{chapter['content']}
            
            检查：逻辑矛盾、时间线错误、人物行为不一致、设定冲突。"""
            
            conflict_report = await self.base_agent.generate_content(prompt)
            if "无冲突" not in conflict_report.lower():
                conflicts.append({
                    'chapter_number': i + 1,
                    'conflict_description': conflict_report
                })
        
        return conflicts
    
    async def suggest_resolutions(self, conflict: Dict, full_context: Dict) -> List[str]:
        prompt = f"""为以下情节冲突提供解决方案：
        冲突描述：{conflict['conflict_description']}
        完整上下文：{full_context}
        
        提供3种不同的解决思路，保持故事连贯性和合理性。"""
        
        resolutions = await self.base_agent.generate_content(prompt)
        return self._parse_resolutions(resolutions)
    
    def _parse_resolutions(self, resolutions: str) -> List[str]:
        return [res.strip() for res in resolutions.split('\n') if res.strip()]

5. 实战案例：科幻短篇创作

async def create_sci_fi_short_story():
    """创建科幻短篇小说的完整示例"""
    system = LiteraryCreationSystem()
    await system.initialize_agents()
    
    try:
        # 创作参数设置
        theme = "人工智能与人类共存的未来社会"
        genre = "科幻"
        target_length = 20000  # 短篇小说字数
        
        print("开始科幻短篇小说创作...")
        print(f"主题: {theme}")
        print(f"类型: {genre}")
        print(f"目标字数: {target_length}")
        print("-" * 50)
        
        # 执行完整创作流程
        result = await system.create_story(theme, genre, target_length)
        
        print("创作完成！")
        print(f"总章节数: {result['final_work']['chapter_count']}")
        print(f"总字数: {result['final_work']['total_word_count']}")
        print("-" * 50)
        
        # 输出第一章内容预览
        first_chapter = result['final_work']['chapters'][0]
        print(f"第一章预览 (前500字):")
        print(first_chapter['content'][:500] + "...")
        
        return result
        
    finally:
        await system.close()

# 运行创作示例
if __name__ == "__main__":
    asyncio.run(create_sci_fi_short_story())

创作质量评估体系

评估维度与指标

评估维度	具体指标	权重	评估方法
情节结构	连贯性、冲突设置、节奏把控	30%	智能体交叉评审
人物塑造	性格一致性、成长弧光、关系网络	25%	人物档案对比分析
文笔质量	语言流畅度、修辞手法、风格统一	20%	文本统计分析
主题表达	主题深度、思想性、情感共鸣	15%	概念一致性检查
创新性	原创性、想象力、独特视角	10%	相似度检测

自动化评估实现

class WritingQualityEvaluator:
    def __init__(self):
        self.metrics_weights = {
            'plot_structure': 0.3,
            'character_development': 0.25,
            'writing_style': 0.2,
            'theme_expression': 0.15,
            'creativity': 0.1
        }
    
    async def evaluate_story(self, story_data: Dict[str, Any], evaluator_agent: CreativeWritingAgent) -> Dict[str, float]:
        evaluation_prompt = self._build_evaluation_prompt(story_data)
        evaluation_result = await evaluator_agent.generate_content(evaluation_prompt)
        
        scores = self._parse_evaluation_scores(evaluation_result)
        weighted_score = self._calculate_weighted_score(scores)
        
        return {
            'detailed_scores': scores,
            'overall_score': weighted_score,
            'evaluation_report': evaluation_result
        }
    
    def _build_evaluation_prompt(self, story_data: Dict) -> str:
        return f"""全面评估以下文学作品质量：
        故事概念：{story_data.get('concept', '')}
        情节结构：{story_data.get('plot', '')}
        人物设定：{story_data.get('characters', '')}
        完整内容：{story_data.get('final_work', {}).get('chapters', [])}
        
        从以下维度评分（0-10分）：
        1. 情节结构：连贯性、冲突设置、节奏把控
        2. 人物塑造：性格一致性、成长弧光、关系网络
        3. 文笔质量：语言流畅度、修辞手法、风格统一
        4. 主题表达：主题深度、思想性、情感共鸣
        5. 创新性：原创性、想象力、独特视角
        
        提供详细评语和改进建议。"""
    
    def _parse_evaluation_scores(self, evaluation_text: str) -> Dict[str, float]:
        # 解析评分结果的实际实现
        return {
            'plot_structure': 8.5,
            'character_development': 7.8,
            'writing_style': 8.2,
            'theme_expression': 8.0,
            'creativity': 7.5
        }
    
    def _calculate_weighted_score(self, scores: Dict[str, float]) -> float:
        return sum(scores[metric] * weight 
                  for metric, weight in self.metrics_weights.items())

最佳实践与优化策略

1. 创作流程优化

2. 性能优化建议

• 批量处理: 对多个章节进行并行创作处理
• 缓存机制: 缓存常用的风格模板和创作模式
• 增量创作: 基于已有内容进行增量式创作，减少重复工作
• 质量阈值: 设置质量评分阈值，自动触发重写机制

3. 成本控制策略

策略类型	具体措施	预期效果
模型选择	根据任务复杂度选择不同规模的模型	降低30-50%API成本
提示优化	精心设计提示词，减少token消耗	减少20-30%token使用
缓存复用

【免费下载链接】autogen 启用下一代大型语言模型应用 项目地址: https://gitcode.com/GitHub_Trending/au/autogen