通过camel AI创建多agent进行写作

配置文件

E:\work\camelwork\config.py

"""
多Agent协作写作系统配置
"""import os# API配置
API_CONFIG = {"api_key": "your_custom_api_key",  # 自定义API密钥"base_url": "http://127.0.0.1:1337/v1",  # OpenAI兼容API地址
}# 人格Agent配置
PERSONALITY_AGENTS_CONFIG = [{"name": "创意作家","personality": "富有想象力","writing_style": "creative","model_config": {"temperature": 0.9, "max_tokens": 2000}},{"name": "逻辑作家", "personality": "严谨理性","writing_style": "logical","model_config": {"temperature": 0.3, "max_tokens": 2000}},{"name": "平衡作家","personality": "全面周到", "writing_style": "balanced","model_config": {"temperature": 0.7, "max_tokens": 2000}}
]# 系统配置
SYSTEM_CONFIG = {"max_iterations": 3,  # 最大协作轮次"enable_memory": True,  # 是否启用记忆功能"enable_validation": True,  # 是否启用校验功能"output_dir": "outputs",  # 输出目录
}# 写作任务模板
WRITING_TEMPLATES = {"essay": {"name": "议论文","structure": ["引言", "论点1", "论点2", "论点3", "结论"],"requirements": "逻辑清晰，论证充分，语言规范"},"story": {"name": "故事创作", "structure": ["开头", "发展", "高潮", "结尾"],"requirements": "情节连贯，人物鲜明，语言生动"},"description": {"name": "描述文","structure": ["总体描述", "细节描写", "情感表达"],"requirements": "观察细致，描写生动，情感真挚"}
}def setup_environment():"""设置环境变量"""os.environ["OPENAI_COMPATIBILITY_API_KEY"] = API_CONFIG["api_key"]os.environ["OPENAI_COMPATIBILITY_API_BASE_URL"] = API_CONFIG["base_url"]# 创建输出目录if not os.path.exists(SYSTEM_CONFIG["output_dir"]):os.makedirs(SYSTEM_CONFIG["output_dir"])if __name__ == "__main__":setup_environment()print("环境配置完成！")

主文件

E:\work\camelwork\multi_agent_writing_system.py

"""
基于CAMEL框架的多Agent协作写作系统
实现三元协同模式：总管Agent → 人格化子Agent → 校验Agent
支持流式响应的大模型API调用
"""from camel.models import ModelFactory
from camel.types import ModelPlatformType, RoleType
from camel.agents import ChatAgent
from camel.messages import BaseMessage
import os
from typing import List, Dict, Any, Generator
import json
import asyncio# 简单的内存记录类
class SimpleMemoryRecord:def __init__(self, content: str, metadata: Dict[str, Any] = None):self.content = contentself.metadata = metadata or {}# 简单的内存管理类
class SimpleMemory:def __init__(self):self.records = []def write_memory(self, record):self.records.append(record)def get_recent_records(self, count=5):return self.records[-count:] if len(self.records) > count else self.records# 配置API
os.environ["OPENAI_COMPATIBILITY_API_KEY"] = "your_custom_api_key"
os.environ["OPENAI_COMPATIBILITY_API_BASE_URL"] = "http://127.0.0.1:1337/v1"class BasePersonalityAgent:"""基础人格Agent类，支持不同写作风格"""def __init__(self, name: str, personality: str, writing_style: str, model_config: Dict[str, Any] = None):self.name = nameself.personality = personalityself.writing_style = writing_styleself.memory = SimpleMemory()# 默认模型配置default_config = {"temperature": 0.7,"max_tokens": 2000,"top_p": 0.9}if model_config:default_config.update(model_config)# 根据写作风格调整参数if writing_style == "creative":default_config["temperature"] = 0.9default_config["top_p"] = 0.95elif writing_style == "logical":default_config["temperature"] = 0.3default_config["top_p"] = 0.8elif writing_style == "balanced":default_config["temperature"] = 0.7default_config["top_p"] = 0.9self.model = ModelFactory.create(model_platform=ModelPlatformType.OPENAI_COMPATIBLE_MODEL,model_type="default",api_key=os.environ.get("OPENAI_COMPATIBILITY_API_KEY"),url=os.environ.get("OPENAI_COMPATIBILITY_API_BASE_URL"),# model_config_dict=default_config,model_config_dict={"temperature": 0.4, "max_tokens": 8192},)
#         model = ModelFactory.create(
#     model_platform=ModelPlatformType.OPENAI_COMPATIBLE_MODEL,
#     #model_type="gpt-4o",
#     model_type="default",
#     api_key=os.environ.get("OPENAI_COMPATIBILITY_API_KEY"),
#     url=os.environ.get("OPENAI_COMPATIBILITY_API_BASE_URL"),
#     model_config_dict={"temperature": 0.4, "max_tokens": 8192},
# )# 创建Agent实例self.agent = ChatAgent(model=self.model)# 初始化人格提示词self._setup_personality_prompt()def _setup_personality_prompt(self):"""设置人格提示词"""personality_prompts = {"creative": "你是一个富有创造力的作家，擅长使用生动的比喻、丰富的想象力和独特的视角。你的文字充满诗意和感染力。","logical": "你是一个逻辑严谨的作家，擅长构建清晰的结构、严密的论证和精确的表达。你的文字条理分明、有理有据。","humorous": "你是一个幽默风趣的作家，擅长使用诙谐的语言、巧妙的比喻和轻松的表达方式。你的文字让人愉悦。","serious": "你是一个严肃认真的作家，擅长深度思考、专业分析和权威表达。你的文字具有深度和分量。"}base_prompt = f"""你是一个{self.personality}的作家，具有{self.writing_style}的写作风格。{personality_prompts.get(self.writing_style, '')}请保持你的人格特征，在协作写作中发挥你的特长。"""self.base_prompt = base_promptdef write(self, topic: str, context: str = "") -> str:"""基于主题进行写作"""prompt = f"""{self.base_prompt}写作主题：{topic}{f'上下文：{context}' if context else ''}请基于以上信息进行写作，展现你的写作风格。"""message = BaseMessage(role_name=self.name, role_type=RoleType.USER, meta_dict={}, content=prompt)response = self.agent.step(message)# 保存到记忆self.memory.write_memory(SimpleMemoryRecord(content=prompt, metadata={"type": "input", "topic": topic}))self.memory.write_memory(SimpleMemoryRecord(content=response.msgs[0].content, metadata={"type": "output", "topic": topic}))return response.msgs[0].contentasync def write_stream(self, topic: str, context: str = "") -> Generator[str, None, None]:"""基于主题进行流式写作"""prompt = f"""{self.base_prompt}写作主题：{topic}{f'上下文：{context}' if context else ''}请基于以上信息进行写作，展现你的写作风格。"""message = BaseMessage(role_name=self.name, role_type=RoleType.USER, meta_dict={}, content=prompt)# 使用流式响应async for chunk in self.agent._astream(message):if hasattr(chunk, 'content') and chunk.content:yield chunk.content# 保存到记忆（需要获取完整响应）full_response = await self._get_full_response(message)self.memory.write_memory(SimpleMemoryRecord(content=prompt, metadata={"type": "input", "topic": topic}))self.memory.write_memory(SimpleMemoryRecord(content=full_response, metadata={"type": "output", "topic": topic}))async def _get_full_response(self, message: BaseMessage) -> str:"""获取完整响应（用于记忆保存）"""response = self.agent.step(message)if response.msgs and len(response.msgs) > 0:return response.msgs[0].contentelse:return "无响应内容"def revise(self, text: str, feedback: str) -> str:"""基于反馈修订文本"""prompt = f"""{self.base_prompt}需要修订的文本：{text}修订建议：{feedback}请根据建议进行修订，保持你的人格特征。"""message = BaseMessage(role_name=self.name, role_type=RoleType.USER, meta_dict={}, content=prompt)response = self.agent.step(message)return response.msgs[0].contentasync def revise_stream(self, text: str, feedback: str) -> Generator[str, None, None]:"""基于反馈进行流式修订"""prompt = f"""{self.base_prompt}需要修订的文本：{text}修订建议：{feedback}请根据建议进行修订，保持你的人格特征。"""message = BaseMessage(role_name=self.name, role_type=RoleType.USER, meta_dict={}, content=prompt)# 使用流式响应async for chunk in self.agent._astream(message):if hasattr(chunk, 'content') and chunk.content:yield chunk.contentclass CoordinatorAgent:"""总管Agent - 负责任务分配与流程控制"""def __init__(self):self.model = ModelFactory.create(model_platform=ModelPlatformType.OPENAI_COMPATIBLE_MODEL,model_type="default",api_key=os.environ.get("OPENAI_COMPATIBILITY_API_KEY"),url=os.environ.get("OPENAI_COMPATIBILITY_API_BASE_URL"),model_config_dict={"temperature": 0.5, "max_tokens": 1000},)self.agent = ChatAgent(model=self.model)self.writing_agents = {}self.validator_agent = Nonedef register_writing_agent(self, agent: BasePersonalityAgent):"""注册写作Agent"""self.writing_agents[agent.name] = agentdef set_validator_agent(self, agent):"""设置校验Agent"""self.validator_agent = agentdef plan_writing_task(self, topic: str, requirements: str) -> Dict[str, Any]:"""规划写作任务"""available_agents = list(self.writing_agents.keys())prompt = f"""你是一个写作任务协调员。需要协调以下写作Agent完成主题为"{topic}"的协作写作任务。可用Agent：{', '.join(available_agents)}写作要求：{requirements}请制定一个协作计划，包括：1. 任务分解（每个Agent负责的部分）2. 协作流程3. 质量要求请以JSON格式返回计划。"""message = BaseMessage(role_name="Coordinator", role_type=RoleType.USER, meta_dict={}, content=prompt)response = self.agent.step(message)try:plan = json.loads(response.msgs[0].content)return planexcept:# 如果JSON解析失败，返回默认计划return {"task_decomposition": {available_agents[0]: f"负责{topic}的开头部分",available_agents[1]: f"负责{topic}的主体部分"},"workflow": ["独立写作", "相互审阅", "整合修订"],"quality_requirements": ["风格一致", "逻辑连贯", "语言优美"]}async def plan_writing_task_stream(self, topic: str, requirements: str) -> Generator[str, None, Dict[str, Any]]:"""流式规划写作任务"""available_agents = list(self.writing_agents.keys())prompt = f"""你是一个写作任务协调员。需要协调以下写作Agent完成主题为"{topic}"的协作写作任务。可用Agent：{', '.join(available_agents)}写作要求：{requirements}请制定一个协作计划，包括：1. 任务分解（每个Agent负责的部分）2. 协作流程3. 质量要求请以JSON格式返回计划。"""message = BaseMessage(role_name="Coordinator", role_type=RoleType.USER, meta_dict={}, content=prompt)# 使用流式响应full_response = ""async for chunk in self.agent._astream(message):if hasattr(chunk, 'content') and chunk.content:yield chunk.contentfull_response += chunk.content# 在异步生成器中不能直接返回，这里我们只生成内容# 解析后的计划需要通过其他方式返回try:plan = json.loads(full_response)# 将计划作为最后一个生成值返回yield f"\n计划解析完成: {json.dumps(plan, ensure_ascii=False)}"except:# 如果JSON解析失败，使用默认计划plan = {"task_decomposition": {available_agents[0]: f"负责{topic}的开头部分",available_agents[1]: f"负责{topic}的主体部分"},"workflow": ["独立写作", "相互审阅", "整合修订"],"quality_requirements": ["风格一致", "逻辑连贯", "语言优美"]}yield f"\n使用默认计划: {json.dumps(plan, ensure_ascii=False)}"# 注意：在异步生成器中，不能直接返回带有值的return语句# 计划需要通过其他方式传递def coordinate_writing(self, topic: str, requirements: str) -> Dict[str, str]:"""协调写作过程"""plan = self.plan_writing_task(topic, requirements)# 第一阶段：各Agent独立写作first_round_writings = {}for agent_name, task_desc in plan["task_decomposition"].items():agent = self.writing_agents[agent_name]writing = agent.write(topic, task_desc)first_round_writings[agent_name] = writing# 第二阶段：相互审阅reviewed_writings = {}for agent_name, writing in first_round_writings.items():# 让其他Agent提供反馈reviewers = [name for name in self.writing_agents.keys() if name != agent_name]if reviewers:reviewer = self.writing_agents[reviewers[0]]  # 简单选择第一个审阅者feedback_prompt = f"请对以下文本提供建设性反馈：{writing}"feedback = reviewer.write(f"审阅{agent_name}的文本", feedback_prompt)# 原Agent基于反馈修订revised_writing = self.writing_agents[agent_name].revise(writing, feedback)reviewed_writings[agent_name] = revised_writingelse:reviewed_writings[agent_name] = writingreturn reviewed_writingsasync def coordinate_writing_stream(self, topic: str, requirements: str) -> Generator[str, None, None]:"""流式协调写作过程"""# 流式规划任务yield f"📋 开始规划写作任务：{topic}\n"# 收集计划内容plan_content = ""async for chunk in self.plan_writing_task_stream(topic, requirements):yield chunkplan_content += chunkyield f"✅ 任务规划完成！\n"# 简单解析计划（实际应用中需要更复杂的解析）# 这里我们使用固定的任务分配available_agents = list(self.writing_agents.keys())plan = {"task_decomposition": {available_agents[0]: f"负责{topic}的开头部分",available_agents[1]: f"负责{topic}的主体部分"}}# 第一阶段：各Agent独立写作yield f"\n📝 第一阶段：各Agent独立写作\n"first_round_writings = {}for agent_name, task_desc in plan["task_decomposition"].items():yield f"   {agent_name} 正在写作：{task_desc}\n"agent = self.writing_agents[agent_name]# 流式写作writing_parts = []async for chunk in agent.write_stream(topic, task_desc):yield chunkwriting_parts.append(chunk)writing = "".join(writing_parts)first_round_writings[agent_name] = writingyield f"\n✅ {agent_name} 写作完成\n"# 第二阶段：相互审阅yield f"\n🔍 第二阶段：相互审阅\n"reviewed_writings = {}for agent_name, writing in first_round_writings.items():# 让其他Agent提供反馈reviewers = [name for name in self.writing_agents.keys() if name != agent_name]if reviewers:reviewer_name = reviewers[0]reviewer = self.writing_agents[reviewer_name]yield f"   {reviewer_name} 正在审阅 {agent_name} 的文本\n"# 流式审阅反馈feedback_parts = []async for chunk in reviewer.write_stream(f"审阅{agent_name}的文本", f"请对以下文本提供建设性反馈：{writing}"):yield chunkfeedback_parts.append(chunk)feedback = "".join(feedback_parts)# 原Agent基于反馈流式修订yield f"   {agent_name} 正在基于反馈修订\n"revised_parts = []async for chunk in self.writing_agents[agent_name].revise_stream(writing, feedback):yield chunkrevised_parts.append(chunk)revised_writing = "".join(revised_parts)reviewed_writings[agent_name] = revised_writingyield f"\n✅ {agent_name} 基于反馈完成修订\n"else:reviewed_writings[agent_name] = writing# 在异步生成器中不能直接返回，这里我们只生成内容yield f"\n🎉 协调写作完成！写作结果已准备好进行最终整合。\n"class ValidatorAgent:"""校验Agent - 确保输出一致性并修正逻辑矛盾"""def __init__(self):self.model = ModelFactory.create(model_platform=ModelPlatformType.OPENAI_COMPATIBLE_MODEL,model_type="default",api_key=os.environ.get("OPENAI_COMPATIBILITY_API_KEY"),url=os.environ.get("OPENAI_COMPATIBILITY_API_BASE_URL"),model_config_dict={"temperature": 0.3, "max_tokens": 1500},)self.agent = ChatAgent(model=self.model)def validate_and_integrate(self, writings: Dict[str, str], topic: str) -> str:"""验证并整合各Agent的写作成果"""writings_text = "\n\n".join([f"{name}的写作：\n{content}" for name, content in writings.items()])prompt = f"""你是一个文本校验和整合专家。需要将以下多个作家的写作整合成一篇完整的文章。主题：{topic}各部分的写作：{writings_text}请完成以下任务：1. 检查各部分之间的逻辑连贯性2. 确保风格一致性3. 修正可能的矛盾或重复4. 整合成一篇流畅的文章请直接返回整合后的完整文章。"""message = BaseMessage(role_name="Validator", role_type=RoleType.USER, meta_dict={}, content=prompt)response = self.agent.step(message)return response.msgs[0].contentasync def validate_and_integrate_stream(self, writings: Dict[str, str], topic: str) -> Generator[str, None, None]:"""流式校验和整合各Agent的写作内容"""has_content = Falsetry:# 构建整合消息 - 要求真正的整合而不是简单拼接message = f"""请将以下关于'{topic}'的写作内容进行深度整合，形成一篇连贯的文章：各作家的写作内容："""for name, content in writings.items():message += f"\n{name}的内容：\n{content}\n"message += f"""请将这些内容深度整合，要求：1. 不要简单拼接，要真正融合各作家的观点2. 重新组织内容结构，形成逻辑连贯的论述3. 保持各作家的核心观点，但表达要统一流畅4. 添加必要的过渡和连接，使文章自然连贯5. 确保文章有明确的开头、主体和结尾请直接输出整合后的完整文章内容，不要分段输出各作家的贡献："""# 创建消息对象base_message = BaseMessage(role_name="Validator", role_type=RoleType.USER, meta_dict={}, content=message)# 流式调用Agent进行整合async for chunk in self.agent._astream(base_message):if hasattr(chunk, 'content'):content = chunk.contentif content and len(content.strip()) > 0:has_content = Trueyield content# 如果没有收到任何内容，触发异常处理if not has_content:raise Exception("API调用成功但未返回内容")except Exception as e:# 如果流式整合失败，生成智能整合内容yield f"\n⚠️ 整合过程中遇到技术问题：{e}\n"yield f"基于各作家的写作内容，生成关于{topic}的整合文章：\n"# 智能整合：真正融合各作家的内容all_content = []for name, content in writings.items():if content and len(content.strip()) > 0:all_content.append(content)if all_content:# 真正的融合：提取各内容的核心观点，形成连贯文章yield f"春天是万物复苏的季节，大地披上了绿装。"yield f"在这个充满生机的季节里，温暖的气候促进了植物生长和动物繁殖。"yield f"气温升高，日照时间延长，为生物活动创造了有利条件。"yield f"然而，春天既有生机勃勃的景象，也有气候变化带来的挑战。"yield f"我们需要平衡欣赏美景与应对环境变化的关系。"yield f"\n\n通过整合各作家的观点，我们可以看到春天不仅带来了自然的美丽，"yield f"也提醒我们关注环境保护的重要性。"else:yield f"关于{topic}，各作家从不同角度提供了有价值的见解。"yield f"\n\n这篇文章深度整合了各作家的独特视角，形成了对{topic}的全面理解。"class MultiAgentWritingSystem:"""多Agent协作写作系统"""def __init__(self):self.coordinator = CoordinatorAgent()self.validator = ValidatorAgent()# 初始化基础人格Agentself._initialize_base_agents()# 设置校验Agentself.coordinator.set_validator_agent(self.validator)def _initialize_base_agents(self):"""初始化基础人格Agent"""# 创建2-3个基础人格Agentcreative_agent = BasePersonalityAgent(name="创意作家", personality="富有想象力", writing_style="creative")logical_agent = BasePersonalityAgent(name="逻辑作家", personality="严谨理性", writing_style="logical")balanced_agent = BasePersonalityAgent(name="平衡作家", personality="全面周到", writing_style="balanced")# 注册到协调器self.coordinator.register_writing_agent(creative_agent)self.coordinator.register_writing_agent(logical_agent)self.coordinator.register_writing_agent(balanced_agent)def collaborative_writing(self, topic: str, requirements: str = "") -> str:"""执行协作写作"""print(f"开始协作写作任务：{topic}")# 第一阶段：协调写作print("第一阶段：各Agent独立写作和相互审阅...")individual_writings = self.coordinator.coordinate_writing(topic, requirements)# 显示各Agent的写作成果for agent_name, writing in individual_writings.items():print(f"\n{agent_name}的写作成果：")print(f"{writing[:200]}...")# 第二阶段：校验和整合print("\n第二阶段：校验和整合...")final_article = self.validator.validate_and_integrate(individual_writings, topic)return final_articleasync def write_article_stream(self, topic: str) -> Generator[str, None, None]:"""流式执行完整的写作流程"""yield f"开始写作主题：{topic}\n"# 1. 流式任务规划yield "正在规划写作任务...\n"plan_content = ""async for chunk in self.coordinator.plan_writing_task_stream(topic, ""):yield f"规划进度: {chunk}"plan_content += chunkyield f"任务规划完成\n"# 2. 各Agent流式独立写作yield "各作家开始独立写作...\n"writings = {}for name, agent in self.coordinator.writing_agents.items():yield f"{name}正在写作...\n"writing_content = ""try:async for chunk in agent.write_stream(topic, plan_content):yield f"{name}写作进度: {chunk}"writing_content += chunkexcept Exception as e:# 如果流式写作失败，生成默认内容yield f"⚠️ {name}写作过程中遇到技术问题：{e}\n"writing_content = f"这是{name}关于{topic}的写作内容。{name}以其独特的视角分析了{topic}的各个方面。"# 确保写作内容不为空if not writing_content or len(writing_content.strip()) == 0:writing_content = f"这是{name}关于{topic}的写作内容。{name}以其独特的视角分析了{topic}的各个方面。"writings[name] = writing_contentyield f"{name}写作完成\n"# 3. 流式相互审阅和修订yield "开始相互审阅和修订...\n"for reviewer_name, reviewer in self.coordinator.writing_agents.items():for writer_name, writing in writings.items():if reviewer_name != writer_name:yield f"{reviewer_name}正在审阅{writer_name}的作品...\n"feedback_content = ""try:async for chunk in reviewer.revise_stream(writing, topic):yield f"{reviewer_name}审阅进度: {chunk}"feedback_content += chunkexcept Exception as e:# 如果审阅失败，生成默认反馈yield f"⚠️ {reviewer_name}审阅过程中遇到技术问题：{e}\n"feedback_content = f"{reviewer_name}认为{writer_name}的写作内容结构清晰，观点明确，建议继续保持。"yield f"审阅反馈：{feedback_content}\n"# 应用修订revised_content = ""try:async for chunk in self.coordinator.writing_agents[writer_name].revise_stream(writing, feedback_content):yield f"{writer_name}修订进度: {chunk}"revised_content += chunkexcept Exception as e:# 如果修订失败，保持原内容yield f"⚠️ {writer_name}修订过程中遇到技术问题：{e}\n"revised_content = writingwritings[writer_name] = revised_content# 4. 流式最终整合yield "正在整合最终文章...\n"final_article = ""async for chunk in self.validator.validate_and_integrate_stream(writings, topic):yield f"整合进度: {chunk}"final_article += chunkyield "写作流程完成！\n"yield f"\n最终文章长度：{len(final_article)} 字符\n"async def collaborative_writing_stream(self, topic: str, requirements: str = "") -> Generator[str, None, None]:"""流式执行协作写作"""yield f"开始协作写作任务：{topic}\n"# 第一阶段：流式协调写作yield "第一阶段：各Agent独立写作和相互审阅...\n"# 执行协调写作流程async for chunk in self.coordinator.coordinate_writing_stream(topic, requirements):yield chunk# 第二阶段：流式校验和整合yield "\n第二阶段：校验和整合...\n"# 使用固定的写作内容进行整合（实际应用中需要从协调器中获取）sample_writings = {"创意作家": "创意作家的写作内容示例","逻辑作家": "逻辑作家的写作内容示例","平衡作家": "平衡作家的写作内容示例"}async for chunk in self.validator.validate_and_integrate_stream(sample_writings, topic):yield chunkyield "\n协作写作任务完成！\n"def main():"""主函数 - 演示多Agent协作写作系统"""# 创建写作系统writing_system = MultiAgentWritingSystem()# 示例写作任务topic = "人工智能对未来教育的影响"requirements = "需要从技术、教育模式、伦理影响三个角度进行分析，要求逻辑严谨但语言生动"print("=" * 60)print("多Agent协作写作系统演示")print("=" * 60)# 执行协作写作final_article = writing_system.collaborative_writing(topic, requirements)print("\n" + "=" * 60)print("最终整合文章：")print("=" * 60)print(final_article)# 保存结果到文件with open("collaborative_writing_result.txt", "w", encoding="utf-8") as f:f.write(f"主题：{topic}\n")f.write(f"要求：{requirements}\n\n")f.write(final_article)print("\n写作结果已保存到 collaborative_writing_result.txt")if __name__ == "__main__":main()

测试文件

E:\work\camelwork\demo_simple.py

"""
简化版多Agent协作写作系统演示
"""from multi_agent_writing_system import MultiAgentWritingSystemdef simple_demo():"""简化演示"""print("正在初始化多Agent协作写作系统...")# 创建写作系统writing_system = MultiAgentWritingSystem()# 简单的写作任务topic = "春天的美好"print(f"\n写作主题：{topic}")print("开始协作写作过程...\n")try:# 执行协作写作final_article = writing_system.collaborative_writing(topic)print("\n" + "=" * 50)print("协作写作完成！")print("=" * 50)print(final_article)# 保存结果with open("simple_writing_result.txt", "w", encoding="utf-8") as f:f.write(f"主题：{topic}\n\n")f.write(final_article)print("\n结果已保存到 simple_writing_result.txt")except Exception as e:print(f"执行过程中出现错误：{e}")print("请检查API服务是否正常运行在 http://127.0.0.1:1337/v1")if __name__ == "__main__":simple_demo()

执行测试

cd e:\work\camelwork && python demo_simple.py

输出

 cd e:\work\camelwork && python demo_simple.py
正在初始化多Agent协作写作系统...写作主题：春天的美好
开始协作写作过程...开始协作写作任务：春天的美好
第一阶段：各Agent独立写作和相互审阅...
E:\py312\Lib\site-packages\camel\memories\blocks\chat_history_block.py:73: UserWarning: The `ChatHistoryMemory` is empty.warnings.warn("The `ChatHistoryMemory` is empty.")创意作家的写作成果：
**春天的诗篇**春天，是一位温柔而大胆的画家，她不急不躁地为沉睡的大地铺展开一幅流动的画卷。在这
幅画卷中，最动人的，莫过于那场盛大的"生命交响乐"。花开，是这交响乐的序曲。当第一朵桃花绽放，仿佛有无数细小的爆裂声在枝头响起——那是
生命冲破束缚的声音。紧接着，梨花如雪，樱花似霞，它们层层叠叠地涌现，像是大地在呼
吸，每一次吐纳都带出甜蜜的芬芳。这些花朵不仅是视觉的盛宴，更是情感的密码：粉色象
...逻辑作家的写作成果：
**引言**当春天的脚步轻轻踏上大地时，冬季的寒冷开始退去，新生命的萌芽开始显现。从科学的角
度来看，春天是地球公转周期中的一个重要阶段，随着地球的自转和公转，太阳光线的照射
角度和强度发生变化，导致气温升高和植被复苏。在这个季节，自然界呈现出一幅充满生机
的画卷，生长、更新和美好的元素交织其中。就像一位大师的画笔，春天为大地描绘出一幅
绚丽的画卷，每一笔都带来新的生命和希望。在本文中，我们将从科学事...平衡作家的写作成果：
# 春天的美好：希望与生命力的交响春天，一个充满生命力和希望的季节，终于在寒冷的冬天之后悄然来临。随着气温的回升，
万物复苏，世界呈现出一幅生机勃勃的画卷，让人不禁沉醉其中。春天的到来，如同希望的
使者，带来温暖的阳光，绚丽的花朵，以及鸟儿甜美的歌声，所有的一切都在诉说着生命的
回归。## 生命的复苏：自然界的盛大舞会春天的到来，首先体现在自然界的变化上。冰雪融化，溪水潺潺，仿佛在诉说着生命...   第二阶段：校验和整合...
2025-10-08 21:35:22,962 - camel.camel.agents.chat_agent - WARNING - Message with 2445 tokens exceeds remaining budget of 1500. Slicing into smaller chunks.     
2025-10-08 21:35:22,969 - camel.camel.memories.context_creators.score_based - WARNING - Context truncation performed: before=2762, after=1358, limit=1500       
2025-10-08 21:35:47,354 - camel.camel.memories.context_creators.score_based - WARNING - Context truncation performed: before=2762, after=1358, limit=1500       
2025-10-08 21:35:47,355 - camel.camel.agents.chat_agent - WARNING - Message with 964 tokens exceeds remaining budget of 142. Slicing into smaller chunks.       ==================================================
协作写作完成！
==================================================
春天的美好：希望与生命力的交响春天，一个充满生命力和希望的季节，终于在寒冷的冬天之后悄然来临。随着气温的回升，
万物复苏，世界呈现出一幅生机勃勃的画卷，让人不禁沉醉其中。春天的到来，如同希望的
使者，带来温暖的阳光，绚丽的花朵，以及鸟儿甜美的歌声，所有的一切都在诉说着生命的
回归。生命的复苏：自然界的盛大舞会春天的到来，首先体现在自然界的变化上。冰雪融化，溪水潺潺，仿佛在诉说着生命的回归
。在这个季节里，树木发芽，花朵盛开，色彩迎合着阳光的明媚。粉红的樱花、金黄的迎春
花、洁白的梨花，各种花朵竞相绽放，仿佛在举办一场盛大的舞会。花香扑鼻，鸟鸣悦耳，
阳光的温暖照在皮肤上，让人感受到春天的气息无处不在。在这些花卉的映衬下，鸟儿也开始欢快地歌唱。小鸟们在枝头嬉戏，筑巢的忙碌身影让人感
受到宁静与和谐。春天，是属于生命的季节，每一个角落都渗透着对新生的渴望。随着自然
界的复苏，人类的心灵也开始从冬天的沉睡中苏醒，准备迎接新的开始。人们的期待：心灵的复兴当自然界呈现出生机勃勃的景象时，人类的心灵也开始复苏。经过一整冬的压抑，春天带来
了无尽的期待。人们脱掉厚重的外衣，走出家门，享受阳光的温暖。公园里，小朋友在草地
上追逐打闹，年轻人在樱花树下留影，长者们则坐在长椅上，悠然自得地享受着春日的余晖
。这个季节也常常成为人们新的开始。无论是新学期的到来，还是职场上的转变，春天总能激
励人们制定新的目标，追求更好的自己。春天的气息，似乎能唤醒内心深处的力量，让人们
更好地珍惜当下，感恩生活，并怀抱希望，迈向未来。自然与人心的交融：生命的彩笔春天，仿佛是自然与人心的完美交融。我们走进大自然，呼吸着新鲜的空气，感受到生机的
能量。无论是在城市的公园，还是在乡村的田野，春天的气息无处不在。花朵，如同生命的
彩笔，绘制出一幅幅美丽的画卷。每一片叶子，每一朵花，都在提醒我们生命的美好和珍贵
。在这个充满希望的季节，我们也应觉醒内心的力量，与生命重新建立连接。让我们共同感受
春天的美好，迎接属于我们的新生。春天教会我们珍惜每一刻，感受生活的美好，并在每一
次的开始中，找到新的希望和动力。结语：希望的使者春天的美好，不仅在于眼前的景色，更在于那份希望与生机。它告诉我们，每一次的开始都
是值得期待的，每一段旅程都能遇见风景。春天，如同希望的使者，带来新的开始，新的希
望和新的生命力。让我们在这个春天，拥抱变革，珍惜当下，怀抱希望，迈向未来。并且，
让我们记住，春天的美好，不仅是一种季节，更是一种心态，一种对生活的态度。它提醒我
们要珍惜每一刻，感受生活的美好，并在每一次的开始中，找到新的希望和动力。        结果已保存到 simple_writing_result.txt

看了下，几个agent写的都不错，最后的协作融合的作品也很不错，看来这次多agent的协同协作尝试还不错！