import time import importlib from core.draft_writer import DraftWriter from core.plot_writer import PlotWriter from core.outline_writer import OutlineWriter from core.writer_utils import KeyPointMsg from core.diff_utils import match_span_by_char import copy import types def load_novel_writer(writer, setting) -> DraftWriter: current_w_name = writer['current_w'] current_w = writer[current_w_name] kwargs = dict( xy_pairs=list(current_w.get('xy_pairs', [['', '']])), model=setting['model'], sub_model=setting['sub_model'], ) kwargs['x_chunk_length'] = current_w['x_chunk_length'] kwargs['y_chunk_length'] = current_w['y_chunk_length'] match current_w_name: case 'draft_w': novel_writer = DraftWriter(**kwargs) case 'outline_w': novel_writer = OutlineWriter(**kwargs) case 'chapters_w' | 'plot_w': novel_writer = PlotWriter(**kwargs) case _: raise ValueError(f"unknown writer: {current_w_name}") return novel_writer def dump_novel_writer(writer, novel_writer, apply_chunks={}, cost=0, currency_symbol='¥'): new_writer = copy.deepcopy(writer) # TODO: dump从设计角度上来说,不应该更改原有的writer,但是在此处copy可能更耗时 current_w_name = new_writer['current_w'] current_w = new_writer[current_w_name] # if current_w_name == 'draft_w': # assert isinstance(novel_writer, DraftWriter), "draft_w需要传入DraftWriter" current_w['xy_pairs'] = list(novel_writer.xy_pairs) current_w['current_cost'] = cost current_w['currency_symbol'] = currency_symbol #current_w['total_cost'] += current_w['current_cost'] current_w['apply_chunks'] = apply_chunks return new_writer def call_write_long_novel(writer, setting): writer = copy.deepcopy(writer) progress = writer['progress'] if not progress or True: progress = dict( cur_op_i = progress['cur_op_i'] if progress else 0, ops = [ { 'before_eval': 'writer["current_w"] = "outline_w"', 'eval': 'call_write(writer, setting, False, "构思全书的大致剧情,并将其以一个故事的形式写下来,只写大致情节。")', 'title': '创作大纲', 'subtitle': '生成大纲' }, { 'eval': 'call_accept(writer, setting)', }, { 'eval': 'call_write(writer, setting, True, "对整个情节进行重写,使其更加有故事性。")', 'title': '创作大纲', 'subtitle': '润色大纲', }, { 'eval': 'call_accept(writer, setting)', }, # 下面是创作剧情 { 'before_eval': 'init_chapters_w(writer)', 'eval': 'call_write(writer, setting, False, "丰富其中的剧情细节。")', 'title': '创作剧情', 'subtitle': '生成剧情' }, { 'eval': 'call_accept(writer, setting)', }, { 'eval': 'call_write(writer, setting, True, "对情节进行重写,使其有更多的剧情细节,同时更加有具有故事性。")', 'title': '创作剧情', 'subtitle': '扩充剧情', }, { 'eval': 'call_accept(writer, setting)', }, # 下面是创作正文 { 'before_eval': 'init_draft_w(writer)', 'eval': 'call_write(writer, setting, False, "创作的是正文,而不是剧情,需要像一个小说家那样去描写这个故事。")', 'title': '创作正文', 'subtitle': '生成正文' }, { 'eval': 'call_accept(writer, setting)', }, { 'eval': 'call_write(writer, setting, True, "润色正文")', 'title': '创作正文', 'subtitle': '润色正文' }, { 'eval': 'call_accept(writer, setting)', } ] ) # TODO: 考虑在init_plot时就给到上下文,类似rewrite_plot title, subtitle = '', '' for op in progress['ops']: if 'title' not in op: op['title'], op['subtitle'] = title, subtitle else: title, subtitle = op['title'], op['subtitle'] writer['progress'] = progress yield writer while progress['cur_op_i'] < len(progress['ops']): current_op = progress['ops'][progress['cur_op_i']] if 'before_eval' in current_op: exec(current_op['before_eval']) writer = yield from eval(current_op['eval']) progress = writer['progress'] progress['cur_op_i'] += 1 yield writer # 当cur_op_i有更新时,也就标志着yield的是一个“稳定版本”的writer_state return writer def match_quote_text(writer, setting, quote_text): novel_writer = load_novel_writer(writer, setting) y_text = novel_writer.y quote_text_span, match_ratio = match_span_by_char(y_text, quote_text) if match_ratio > 0.5: aligned_span, _ = novel_writer.align_span(y_span=quote_text_span) return aligned_span, y_text[aligned_span[0]:aligned_span[1]] else: return None, '' # 这是后端函数,接受前端writer_state的copy做为输入 # 返回的是修改后的writer_state,注意yield的值一般被用于前端展示执行的过程和进度 # 只有return值才会被前端考虑用于writer_state的更新 def call_write(writer, setting, auto_write=False, suggestion=None): novel_writer = load_novel_writer(writer, setting) current_w = writer[writer['current_w']] current_w['xy_pairs'] = list(novel_writer.xy_pairs) quote_span = writer['quote_span'] if auto_write: assert quote_span is None, "auto_write模式下,不能有quote_text" generator = novel_writer.auto_write() else: # TODO: writer.write 应该保证无论什么prompt,都能够同时适应y为空和y有值地情况 # 换句话说,就是虽然可以单列出一个“新建正文”,但用扩写正文也能实现同样的效果。 generator = novel_writer.write(suggestion, y_span=quote_span) prompt_outputs = [] for kp_msg in generator: if isinstance(kp_msg, KeyPointMsg): # 如果要支持关键节点保存,需要计算一个编辑上的更改,然后在这里yield writer yield kp_msg continue else: chunk_list = kp_msg current_cost = 0 apply_chunks = [] prompt_outputs.clear() for output, chunk in chunk_list: prompt_outputs.append(output) current_text = "" current_cost += output['response_msgs'].cost currency_symbol = output['response_msgs'].currency_symbol cost_info = f"\n(预计花费:{output['response_msgs'].cost:.4f}{output['response_msgs'].currency_symbol})" if 'plot2text' in output: current_text += f"正在建立映射关系..." + cost_info + '\n' else: current_text += output['text'] + cost_info + '\n' apply_chunks.append((chunk, 'y_chunk', current_text)) new_writer = dump_novel_writer(writer, novel_writer, apply_chunks=apply_chunks, cost=current_cost, currency_symbol=currency_symbol) new_writer['prompt_outputs'] = prompt_outputs yield new_writer # 这里是计算出一个编辑上的更改,方便前端显示,后续diff功能将不由writer提供,因为这是为了显示的要求 apply_chunks = [] for chunk, key, value in load_novel_writer(writer, setting).diff_to(novel_writer): apply_chunks.append((chunk, key, value)) writer[writer['current_w']]['apply_chunks'] = apply_chunks writer['prompt_outputs'] = prompt_outputs return writer def call_accept(writer, setting): current_w_name = writer['current_w'] current_w = writer[current_w_name] novel_writer = load_novel_writer(writer, setting) for chunk, key, text in current_w['apply_chunks']: novel_writer.apply_chunk(chunk, key, text) writer = dump_novel_writer(writer, novel_writer) return writer