大模型学习笔记十三：工作流

一、了解工作流对大模型的辅助意义

1）那些影响大模型应用的效果

①模型能力：

    - 通识理解和泛化能力- 输入信息理解、推理、规划、执行能力- 输入信息补充知识学习能力- 文字生成创作的风格

②模型输出控制

    -1)单次请求控制- Prompt表达优化- 以CoT为代表的思维链控制方法- 输出格式控制（文本格式语法、工程结构化数据输出…）-2)多次请求控制- 以ReAct（Action-Observation-Reflection）为代表的多轮自我反思优化- 复杂任务的执行过程编排管理

2）单词请求的局限性

① 上下文窗口长度限制、输出长度限制（早期的LangChain长文本Summarize）
②直接进行CoT控制（尤其是用自然语言表达CoT，也就是思维链）会输出思考过程，但我们不希望用户看到这个过程
③ 随着工作进展出现的新信息，对任务时序、编排有依赖的信息，不一定能在单次请求中一次性完成输入

3）工作流优势（将工作拆分成多个工作节点）

①将工作任务拆分成多个工作节点
②能够将模型单次请求调用视作一个工作节点
③能够灵活将其他代码逻辑也写入工作节点
④能够对工作节点进行任务编排
⑤能够在工作节点之间进行数据传递

二、学会复现吴恩达的翻译工作流开源项目

1）项目要求和原理解析

• 项目介绍
  翻译文本
• 项目地址：
  https://github.com/andrewyng/translation-agent
• 项目原理
  让模型在完成首轮翻译之后，通过自我反思后修正的工作流优化翻译结果，以提升最终文本翻译的质量
• 关键代码文件
  https://github.com/andrewyng/translation-agent/blob/main/src/translation_agent/utils.py
• 关键步骤：

1）第一步：
输入信息：原始文本语言(source_lang) 、翻译目标语言(target_lang) 和 原始文本(source_text)
角色设定：以翻译文本为任务目标的语言学家
输出结果：基于所有输入信息，对 原始文本(source_text) 进行 **第一轮翻译的结果(translation_1)**；2）第二步：
输入信息：原始文本语言(source_lang) 、翻译目标语言(target_lang) 、 原始文本(source_text) 和 第一轮翻译结果(translation_1)
角色设定：以阅读原始文本和翻译文本，并给出翻译改进意见为任务目标的语言学家
输出结果：基于所有输入信息，对 第一轮翻译结果(translation_1) 提出的 改进意见反思(reflection)3）第三步：
输入信息：原始文本语言(source_lang) 、翻译目标语言(target_lang) 、 原始文本(source_text) 、 第一轮翻译结果(translation_1) 和 改进意见反思(reflection)
角色设定：以翻译文本为任务目标的语言学家（和第一步相同）
输出结果：基于所有输入信息，给出的第二轮优化后翻译结果(translation_2)

• 代码讲解(主函数one_chunk_translate_text)
  1）总体先通过one_chunk_translate_text获得翻译后的第一次翻译结果文本
  2）再通过one_chunk_reflect_on_translation获得反省建议
  3）再根据建议和第一次翻译的结果优化结果

def one_chunk_translate_text(source_lang: str, target_lang: str, source_text: str, country: str = ""
) -> str:"""Translate a single chunk of text from the source language to the target language.This function performs a two-step translation process:1. Get an initial translation of the source text.2. Reflect on the initial translation and generate an improved translation.Args:source_lang (str): The source language of the text.target_lang (str): The target language for the translation.source_text (str): The text to be translated.country (str): Country specified for target language.Returns:str: The improved translation of the source text."""translation_1 = one_chunk_initial_translation(source_lang, target_lang, source_text)reflection = one_chunk_reflect_on_translation(source_lang, target_lang, source_text, translation_1, country)translation_2 = one_chunk_improve_translation(source_lang, target_lang, source_text, translation_1, reflection)return translation_2

2）使用langGraph复现这个工作流

• 代码讲解

import json
import openai
from ENV import deep_seek_url, deep_seek_api_key, deep_seek_default_model
from langgraph.graph import StateGraph, START, END
import os# 模型请求准备
client = openai.OpenAI(api_key = deep_seek_api_key,base_url =deep_seek_url
)
default_model = deep_seek_default_modeldef get_completion(prompt: str,system_message: str = "You are a helpful assistant.",model: str = default_model,temperature: float = 0.3,json_mode: bool = False,
):response = client.chat.completions.create(model=model,temperature=temperature,top_p=1,messages=[{"role": "system", "content": system_message},{"role": "user", "content": prompt},],)return response.choices[0].message.content# 定义传递的信息结构
from typing import TypedDict, Optional
class State(TypedDict):source_lang: strtarget_lang: str   #必须的参数source_text: strcountry: Optional[str] = Nonetranslation_1: Optional[str] = Nonereflection: Optional[str] = Nonetranslation_2: Optional[str] = None# 创建一个工作流对象
workflow = StateGraph(State)# 定义三个工作块
"""
获取state中的信息：state.get("key_name")
更新state中的信息：return { "key_name": new_value }
"""
def initial_translation(state):source_lang = state.get("source_lang")target_lang = state.get("target_lang")source_text = state.get("source_text")system_message = f"You are an expert linguist, specializing in translation from {source_lang} to {target_lang}."prompt = f"""This is an {source_lang} to {target_lang} translation, please provide the {target_lang} translation for this text. \
Do not provide any explanations or text apart from the translation.
{source_lang}: {source_text}{target_lang}:"""translation = get_completion(prompt, system_message=system_message)print("[初次翻译结果]: \n", translation)return { "translation_1": translation }def reflect_on_translation(state):source_lang = state.get("source_lang")target_lang = state.get("target_lang")source_text = state.get("source_text")country = state.get("country") or ""translation_1 = state.get("translation_1")system_message = f"You are an expert linguist specializing in translation from {source_lang} to {target_lang}. \
You will be provided with a source text and its translation and your goal is to improve the translation."additional_rule = (f"The final style and tone of the translation should match the style of {target_lang} colloquially spoken in {country}."if country != ""else "")prompt = f"""Your task is to carefully read a source text and a translation from {source_lang} to {target_lang}, and then give constructive criticism and helpful suggestions to improve the translation. \
{additional_rule}The source text and initial translation, delimited by XML tags <SOURCE_TEXT></SOURCE_TEXT> and <TRANSLATION></TRANSLATION>, are as follows:<SOURCE_TEXT>
{source_text}
</SOURCE_TEXT><TRANSLATION>
{translation_1}
</TRANSLATION>When writing suggestions, pay attention to whether there are ways to improve the translation's \n\
(i) accuracy (by correcting errors of addition, mistranslation, omission, or untranslated text),\n\
(ii) fluency (by applying {target_lang} grammar, spelling and punctuation rules, and ensuring there are no unnecessary repetitions),\n\
(iii) style (by ensuring the translations reflect the style of the source text and takes into account any cultural context),\n\
(iv) terminology (by ensuring terminology use is consistent and reflects the source text domain; and by only ensuring you use equivalent idioms {target_lang}).\n\Write a list of specific, helpful and constructive suggestions for improving the translation.
Each suggestion should address one specific part of the translation.
Output only the suggestions and nothing else."""reflection = get_completion(prompt, system_message=system_message)print("[初次翻译结果]: \n", reflection)return { "reflection": reflection }def improve_translation(state):source_lang = state.get("source_lang")target_lang = state.get("target_lang")source_text = state.get("source_text")translation_1 = state.get("translation_1")reflection = state.get("reflection")system_message = f"You are an expert linguist, specializing in translation editing from {source_lang} to {target_lang}."prompt = f"""Your task is to carefully read, then edit, a translation from {source_lang} to {target_lang}, taking into
account a list of expert suggestions and constructive criticisms.The source text, the initial translation, and the expert linguist suggestions are delimited by XML tags <SOURCE_TEXT></SOURCE_TEXT>, <TRANSLATION></TRANSLATION> and <EXPERT_SUGGESTIONS></EXPERT_SUGGESTIONS> \
as follows:<SOURCE_TEXT>
{source_text}
</SOURCE_TEXT><TRANSLATION>
{translation_1}
</TRANSLATION><EXPERT_SUGGESTIONS>
{reflection}
</EXPERT_SUGGESTIONS>Please take into account the expert suggestions when editing the translation. Edit the translation by ensuring:(i) accuracy (by correcting errors of addition, mistranslation, omission, or untranslated text),
(ii) fluency (by applying {target_lang} grammar, spelling and punctuation rules and ensuring there are no unnecessary repetitions), \
(iii) style (by ensuring the translations reflect the style of the source text)
(iv) terminology (inappropriate for context, inconsistent use), or
(v) other errors.Output only the new translation and nothing else."""translation_2 = get_completion(prompt, system_message)print("[初次翻译结果]: \n", translation_2)return { "translation_2": translation_2 }# 规划执行任务
## 节点（node）注册
workflow.add_node("initial_translation", initial_translation)
workflow.add_node("reflect_on_translation", reflect_on_translation)
workflow.add_node("improve_translation", improve_translation)
## 连接节点
workflow.set_entry_point("initial_translation")
#workflow.add_edge(START, )
workflow.add_edge("initial_translation", "reflect_on_translation")
workflow.add_edge("reflect_on_translation", "improve_translation")
workflow.add_edge("improve_translation", END)# 开始执行
app = workflow.compile()
result = app.invoke({"source_lang": "English","target_lang": "中文","source_text": """Translation Agent: Agentic translation using reflection workflow
This is a Python demonstration of a reflection agentic workflow for machine translation. The main steps are:Prompt an LLM to translate a text from source_language to target_language;
Have the LLM reflect on the translation to come up with constructive suggestions for improving it;
Use the suggestions to improve the translation.
Customizability
By using an LLM as the heart of the translation engine, this system is highly steerable. For example, by changing the prompts, it is easier using this workflow than a traditional machine translation (MT) system to:Modify the output's style, such as formal/informal.
Specify how to handle idioms and special terms like names, technical terms, and acronyms. For example, including a glossary in the prompt lets you make sure particular terms (such as open source, H100 or GPU) are translated consistently.
Specify specific regional use of the language, or specific dialects, to serve a target audience. For example, Spanish spoken in Latin America is different from Spanish spoken in Spain; French spoken in Canada is different from how it is spoken in France.
This is not mature software, and is the result of Andrew playing around with translations on weekends the past few months, plus collaborators (Joaquin Dominguez, Nedelina Teneva, John Santerre) helping refactor the code.According to our evaluations using BLEU score on traditional translation datasets, this workflow is sometimes competitive with, but also sometimes worse than, leading commercial offerings. However, we’ve also occasionally gotten fantastic results (superior to commercial offerings) with this approach. We think this is just a starting point for agentic translations, and that this is a promising direction for translation, with significant headroom for further improvement, which is why we’re releasing this demonstration to encourage more discussion, experimentation, research and open-source contributions.If agentic translations can generate better results than traditional architectures (such as an end-to-end transformer that inputs a text and directly outputs a translation) -- which are often faster/cheaper to run than our approach here -- this also provides a mechanism to automatically generate training data (parallel text corpora) that can be used to further train and improve traditional algorithms. (See also this article in The Batch on using LLMs to generate training data.)Comments and suggestions for how to improve this are very welcome!"""
})print(result)

3）使用Agently Workflow分别复现这个工作流

三、了解构成工作流系统的关键元素

1）基本要素

2）大模型应用工作流需要具备的特性

3）LangGraph的工作流要素图示

4）Agently Workflow的工作流要素图示

5）LangGraph和Agently Workflow的能力对比表

四、学会构建一个更复杂的故事工作流

1）设计思路

2）实现方案

3）进一步思考和讨论

4）其他信息

Mermaid在线渲染网站：mermaid.live
手绘风格流程图在线编辑：excalidraw.com

五、备注

# 因为使用的langgraph可能需要依赖langchain 0.2.10版本，但其他组件依赖langchain 0.1.20版本
# 使用后需要对langchain进行降级，以免在其他代码使用出现运行错误
!pip install langchain==0.1.20
!pip install langchain-openai==0.1.6
!pip install langchain-community==0.0.38