diff --git a/config/translation_config.yaml b/config/translation_config.yaml
new file mode 100644
index 000000000..a156cd8a7
--- /dev/null
+++ b/config/translation_config.yaml
@@ -0,0 +1,18 @@
+# Example YAML configuration for TranslationDataGenerator
+# You can use this file with TranslationDataGenerator.from_yaml('config/translation_config.yaml')
+
+base_url: "http://localhost:11434/v1"
+api_key: ""
+init_translate_model: "gpt-oss:latest"
+reflection_model: "gpt-oss:latest"
+improvement_model: "gpt-oss:latest"
+hf_tokenizer: "openai/gpt-oss-20b"
+hf_token: ""
+max_token_per_chunk: 5000
+temperature: 1.0
+top_p: 1.0
+stop: ["<|return|>","<|endoftext|>", "<|call|>"]
+max_tokens: 24576
+source_lang: "English"
+target_lang: "Traditional Chinese"
+country: "Taiwan"
\ No newline at end of file
diff --git a/docs/curate-text/generate-data/pipelines/translation.md b/docs/curate-text/generate-data/pipelines/translation.md
new file mode 100644
index 000000000..eecdccd42
--- /dev/null
+++ b/docs/curate-text/generate-data/pipelines/translation.md
@@ -0,0 +1,219 @@
+---
+description: "Translate text documents using a multi-step LLM pipeline with reflection and improvement"
+categories: ["how-to-guides"]
+tags: ["translation", "llm", "synthetic-data", "reflection", "improvement"]
+personas: ["data-scientist-focused", "mle-focused"]
+difficulty: "intermediate"
+content_type: "how-to"
+modality: "text-only"
+---
+
+(text-gen-data-pipelines-translation)=
+# Translation Pipeline
+
+This pipeline translates text documents using a multi-step process with large language models (LLMs), including initial translation, reflection, and improvement. It leverages the NeMo Curator framework and supports both function-based and YAML-based configuration for flexible workflows.
+
+## Before You Start
+
+- **LLM Client Setup**: The `TranslationDataGenerator` requires an `OpenAIClient` instance to interface with language models. See the [LLM services documentation](text-generate-data-connect-service) for details on configuring your client and model provider.
+
+---
+
+## Setup Steps
+
+### Set up the LLM Client
+
+Configure your LLM client (example with OpenAI):
+
+```python
+from openai import OpenAI
+
+openai_client = OpenAI(
+    base_url="http://localhost:8000/v1",
+    api_key="<insert API key>"
+)
+```
+
+### Create the NeMo Curator Client Wrapper
+
+Wrap the client with NeMo Curator's client wrapper:
+
+```python
+from nemo_curator import OpenAIClient
+
+client = OpenAIClient(openai_client)
+```
+
+### Initialize the Translation Generator
+
+Create the `TranslationDataGenerator` instance:
+
+```python
+from nemo_curator.synthetic.translate import TranslationDataGenerator
+
+# Create a TranslationDataGenerator instance with specified parameters
+generator = TranslationDataGenerator(
+    base_url="http://localhost:11434/v1",                   # (Change this) Base URL for local API (P.S: Ollama supports the OpenAI API format.)
+    api_key="",                                             # API key (empty if not required)
+    init_translate_model="gpt-oss:latest",                  # Initial translation model
+    reflection_model="gpt-oss:latest",                      # Reflection model for improvement
+    improvement_model="gpt-oss:latest",                     # Model for translation improvement
+    hf_tokenizer="openai/gpt-oss-20b" ,                     # (Change this) HuggingFace model for tokenization
+    hf_token=None,                                            # (Change this) HuggingFace authentication token
+    temperature=1.0,                                        # Sampling temperature for generation
+    top_p=1.0,                                              # Nucleus sampling parameter
+    max_tokens=24576,                                        # Maximum tokens for input
+    stop=["<|return|>","<|endoftext|>", "<|call|>"],        # (Change this) Stop TOKEN sequences
+    max_token_per_chunk=5000,                               # Max tokens per chunk for translation
+    source_lang="English",                                  # Source language
+    target_lang="Traditional Chinese",                      # Target language
+    country="Taiwan",                                       # (Optional) Country context for translation
+)
+```
+
+### YAML-Based Configuration
+
+You can also configure the generator using a YAML file:
+
+```python
+# config/translation_config.yaml
+# See the provided example in the repository
+
+generator_yaml = TranslationDataGenerator.from_yaml("config/translation_config.yaml")
+```
+
+---
+
+## Translation Workflow
+
+### Translate a Single Text
+
+```python
+text = "Once upon a time, there were three little pig brothers..."
+translations = generator.generate(text)
+print(generator.parse_response(translations))
+```
+
+### Translate Using YAML Configuration
+
+```python
+translations = generator_yaml.generate_from_yaml("config/translation_config.yaml", text)
+print(translations)
+```
+
+### Batch Translation with DataFrames
+
+Efficiently translate multiple texts in a pandas DataFrame:
+
+```python
+import pandas as pd
+
+df = pd.DataFrame({
+    "text": [
+        "Once upon a time, there were three little pig brothers...",
+        "The quick brown fox jumps over the lazy dog."
+    ]
+})
+
+df_translated = generator_yaml.generate_from_dataframe(df, text_column="text", batch_size=16)
+print(df_translated.head())
+```
+
+### Asynchronous Batch Translation
+
+For large-scale translation tasks, use the async pipeline:
+
+```python
+import asyncio
+
+async def async_translate():
+    df_translated = await generator_yaml.async_generate_from_dataframe(df, text_column="text", batch_size=16)
+    print("[Async]", df_translated.head())
+
+asyncio.run(async_translate())
+```
+
+---
+
+## Pipeline Steps Explained
+
+1. **Initial Translation**: The input text is translated using the specified LLM model.
+2. **Reflection**: The initial translation is reviewed and refined, optionally with country-specific context.
+3. **Improvement**: The translation is finalized using feedback from the reflection step.
+
+This multi-step approach improves translation quality and contextual accuracy.
+
+---
+
+## Example YAML Configuration
+
+```yaml
+base_url: "http://localhost:8000/v1"
+api_key: ""
+init_translate_model: "gpt-oss:latest"
+reflection_model: "gpt-oss:latest"
+improvement_model: "gpt-oss:latest"
+hf_tokenizer: "openai/gpt-oss-20b"
+hf_token: ""
+max_token_per_chunk: 5000
+temperature: 1.0
+top_p: 1.0
+max_tokens: 8192
+source_lang: "English"
+target_lang: "Traditional Chinese"
+country: "Taiwan"
+```
+
+---
+
+## Dataset Translation Example (using HuggingFace Datasets as a template)
+
+### Initial TranslationDataGenerator
+
+```python
+# (Optional) Import BaseSettings from pydantic for configuration management
+from pydantic.v1 import BaseSettings
+
+# (Optional) Define a Settings class to store model and API configuration
+class Settings(BaseSettings):
+    hf_token: str = None                            # (Change this) HuggingFace token for authentication
+    hf_model: str = "openai/gpt-oss-20b"            # (Change this) HuggingFace model for tokenization
+    model_name: str = "gpt-oss:latest"              # (Change this) Local model name
+    base_url: str = "http://localhost:11434/v1"     # (Change this) Base URL for local API (P.S: Ollama supports the OpenAI API format.)
+
+# Instantiate the Settings object to access configuration
+setting = Settings()
+
+# Import the TranslationDataGenerator for synthetic translation tasks
+from nemo_curator.synthetic.translate import TranslationDataGenerator
+
+# Create a TranslationDataGenerator instance with specified parameters
+generator = TranslationDataGenerator(
+    base_url=setting.base_url,                              # API endpoint
+    api_key="",                                             # API key (empty if not required)
+    init_translate_model=setting.model_name,                # Initial translation model
+    reflection_model=setting.model_name,                    # Reflection model for improvement
+    improvement_model=setting.model_name,                   # Model for translation improvement
+    hf_tokenizer=setting.hf_model,                          # Tokenizer model from HuggingFace
+    hf_token=setting.hf_token,                              # HuggingFace authentication token
+    temperature=1.0,                                        # Sampling temperature for generation
+    top_p=1.0,                                              # Nucleus sampling parameter
+    max_tokens=24576,                                        # Maximum tokens for input
+    stop=["<|return|>","<|endoftext|>", "<|call|>"],        # Stop TOKEN sequences
+    max_token_per_chunk=5000,                               # Max tokens per chunk for translation
+    source_lang="English",                                  # Source language
+    target_lang="Traditional Chinese",                      # Target language
+    country="Taiwan",                                       # (Optional) Country context for translation
+)
+```
+
+### 
+
+---
+
+## References
+
+- [NeMo Curator Documentation](https://github.com/NVIDIA/NeMo-Curator)
+- [Translation Agent Reference](https://github.com/andrewyng/translation-agent)
+
+---
diff --git a/examples/README.md b/examples/README.md
index 0e1055270..70ba7de3b 100644
--- a/examples/README.md
+++ b/examples/README.md
@@ -20,6 +20,7 @@ These include:
 | raw_download_common_crawl.py          | Download the raw compressed WARC files from Common Crawl without extracting them.                                |
 | semdedup_example.py                   | Use the `SemDedup` class to perform semantic deduplication on text data.                                         |
 | task_decontamination.py               | Remove segments of downstream evaluation tasks from a dataset.                                                   |
+| translation.py                        | Language/region translation using `TranslationDataGenerator` for any Base-LLM supported language.                |
 
 Before running any of these scripts, we strongly recommend displaying `python <script name>.py --help` to ensure that any needed or relevant arguments are specified.
 
diff --git a/examples/datasets_translation.ipynb b/examples/datasets_translation.ipynb
new file mode 100644
index 000000000..9623c9c1d
--- /dev/null
+++ b/examples/datasets_translation.ipynb
@@ -0,0 +1,200 @@
+{
+ "cells": [
+  {
+   "cell_type": "markdown",
+   "id": "aa8d1a83",
+   "metadata": {},
+   "source": [
+    "# Dataset Translation Example (using HuggingFace Datasets as a template)"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "id": "b4d96ca7",
+   "metadata": {},
+   "source": [
+    "## Initial TranslationDataGenerator"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "6085b2f8",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "# (Optional) Import BaseSettings from pydantic for configuration management\n",
+    "from pydantic.v1 import BaseSettings\n",
+    "\n",
+    "# (Optional) Define a Settings class to store model and API configuration\n",
+    "class Settings(BaseSettings):\n",
+    "    hf_token: str = None                        # (Change this) HuggingFace token for authentication\n",
+    "    hf_model: str = \"openai/gpt-oss-20b\"        # (Change this) HuggingFace model for tokenization\n",
+    "    model_name: str = \"gpt-oss:latest\"          # (Change this) Local model name\n",
+    "    base_url: str = \"http://localhost:11434/v1\" # (Change this) Base URL for local API (P.S: Ollama supports the OpenAI API format.)\n",
+    "\n",
+    "# Instantiate the Settings object to access configuration\n",
+    "setting = Settings()\n",
+    "\n",
+    "# Import the TranslationDataGenerator for synthetic translation tasks\n",
+    "from nemo_curator.synthetic.translate import TranslationDataGenerator\n",
+    "\n",
+    "# Create a TranslationDataGenerator instance with specified parameters\n",
+    "generator = TranslationDataGenerator(\n",
+    "    base_url=setting.base_url,                              # API endpoint\n",
+    "    api_key=\"\",                                             # API key (empty if not required)\n",
+    "    init_translate_model=setting.model_name,                # Initial translation model\n",
+    "    reflection_model=setting.model_name,                    # Reflection model for improvement\n",
+    "    improvement_model=setting.model_name,                   # Model for translation improvement\n",
+    "    hf_tokenizer=setting.hf_model,                          # Tokenizer model from HuggingFace\n",
+    "    hf_token=setting.hf_token if setting.hf_token != \"\" else None, # HuggingFace authentication token\n",
+    "    temperature=1.0,                                        # Sampling temperature for generation\n",
+    "    top_p=1.0,                                              # Nucleus sampling parameter\n",
+    "    max_tokens=24576,                                        # Maximum tokens for input\n",
+    "    stop=[\"<|return|>\",\"<|endoftext|>\", \"<|call|>\"],        # Stop TOKEN sequences\n",
+    "    max_token_per_chunk=5000,                               # Max tokens per chunk for translation\n",
+    "    source_lang=\"English\",                                  # Source language\n",
+    "    target_lang=\"Traditional Chinese\",                      # Target language\n",
+    "    country=\"Taiwan\",                                       # (Optional) Country context for translation\n",
+    ")"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "id": "03163ec1",
+   "metadata": {},
+   "source": [
+    "## Load & Translate Sample Dataset"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 2,
+   "id": "50379d6f",
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "Processing example 1/1...\n",
+      "EXTRACTED: 3. (6 分) 一家建設公司正在建造隧道。當隧道完成了三分之一時，他們開始使用新設備，該設備將施工速度提升 20%，並將工作時間削減至原來的 80%。最終完成隧道共花費 185 天。若不使用新設備，繼續以原速施工，完成隧道仍需 ______ 天。\n"
+     ]
+    }
+   ],
+   "source": [
+    "# Import required libraries for data processing and translation\n",
+    "import pandas as pd\n",
+    "from datasets import load_dataset\n",
+    "import re\n",
+    "import json\n",
+    "\n",
+    "# Load the Sample Dataset from HuggingFace using the provided token\n",
+    "ds = load_dataset(\"TsukiOwO/Translation-Sample-Dataset\", \"default\", token=setting.hf_token if setting.hf_token != \"\" else None)[\"train\"]\n",
+    "messages_zh_all = []    # List to store translated message sets\n",
+    "max_examples = 1        # (Optional) Limit the number of examples processed\n",
+    "\n",
+    "# Iterate through filtered examples and translate messages\n",
+    "for idx, example in enumerate(ds):\n",
+    "    # (Optional) Set a Limit for example\n",
+    "    if idx >= max_examples:\n",
+    "        break\n",
+    "    \n",
+    "    print(f\"Processing example {idx + 1}/{max_examples}...\") \n",
+    "\n",
+    "    messages = example['messages']  # Get the list of messages for the example\n",
+    "    messages_zh = []  # List to store translated messages for this example\n",
+    "\n",
+    "    for msg in messages:\n",
+    "        role = msg.get('role')  # Get the role (user or assistant)\n",
+    "        content = msg.get('content', '')  # Get the message content\n",
+    "\n",
+    "        if role == 'user':\n",
+    "            # Translate user message content\n",
+    "            translations = generator.generate(content, debug=True) # (Optional) debug=True: Can get the `extract_content` step information \n",
+    "            translated = generator.parse_response(translations)\n",
+    "            messages_zh.append({\"role\": role, \"content\": translated})\n",
+    "\n",
+    "        elif role == 'assistant':\n",
+    "            # Extract and translate <think> content if present\n",
+    "            think_match = re.search(r\"<think>(.*?)</think>\", content, re.DOTALL)\n",
+    "            if think_match:\n",
+    "                think_text = think_match.group(1).strip()\n",
+    "                translations = generator.generate(think_text)\n",
+    "                think_translated = generator.parse_response(translations)\n",
+    "            else:\n",
+    "                think_translated = \"\"\n",
+    "            \n",
+    "            # Extract and translate non-<think> content\n",
+    "            result_text = re.sub(r\"<think>.*?</think>\", \"\", content, flags=re.DOTALL).strip()\n",
+    "            translations = generator.generate(result_text)\n",
+    "            result_translated = generator.parse_response(translations)\n",
+    "            # Merge translated <think> and non-<think> content\n",
+    "            combined_content = f\"<think>\\n{think_translated}</think>\\n\\n{result_translated}\"\n",
+    "            messages_zh.append({\"role\": \"assistant\", \"content\": combined_content})\n",
+    "            messages_zh_all.append(messages_zh)  # Add translated messages for this example\n",
+    "\n",
+    "# Create a DataFrame to store all translated message sets\n",
+    "df = pd.DataFrame()\n",
+    "df['messages_zh'] = messages_zh_all"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "id": "a4dcbcce",
+   "metadata": {},
+   "source": [
+    "## Show The Result"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 3,
+   "id": "031e50ec",
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "[\n",
+      "  {\n",
+      "    \"role\": \"user\",\n",
+      "    \"content\": \"3. (6 分) 一家建設公司正在建造隧道。當隧道完成了三分之一時，他們開始使用新設備，該設備將施工速度提升 20%，並將工作時間削減至原來的 80%。最終完成隧道共花費 185 天。若不使用新設備，繼續以原速施工，完成隧道仍需 ______ 天。\"\n",
+      "  },\n",
+      "  {\n",
+      "    \"role\": \"assistant\",\n",
+      "    \"content\": \"<think>\\n先來分析這個問題。\\n\\n一家建設公司正在建造一條隧道。完成隧道的三分之一後，他們改用新設備。該設備使施工速度提高 20%，但每日工作時數被削減至原來的 80%。使用新設備後，總共耗時 185 天。請問若不使用新設備，改以原始速度繼續建造，總共會耗時多少天？\\n\\n設全長為 1 個隧道。  \\n原始施工速率為 \\\\(R\\\\)（隧道/天），若不變則需完成整條隧道的時間為  \\n\\\\[\\nT=\\\\frac{1}{R}\\\\quad(\\\\text{天}) .\\n\\\\]\\n\\n**第一階段**  \\n以原速率完成三分之一，所需時間  \\n\\\\[\\nt_1=\\\\frac{\\\\tfrac13}{R}=\\\\frac{1}{3R}\\\\quad(\\\\text{天}) .\\n\\\\]\\n\\n**第二階段**  \\n剩餘三分之二按新設備施工。新設備使速率提升 20%，即從 \\\\(R\\\\) 變為 \\\\(1.2R\\\\)。  \\n原本每日工作 \\\\(H\\\\) 小時，提升後速率為每小時 \\\\(\\\\tfrac{R}{H}\\\\)。提高 20% 之後為 \\\\(1.2\\\\tfrac{R}{H}\\\\)。  \\n但每日工作時數下降至 \\\\(0.8H\\\\)，因此每日新速率為  \\n\\\\[\\n1.2\\\\tfrac{R}{H}\\\\times 0.8H = 0.96R \\\\quad(\\\\text{隧道/天}) .\\n\\\\]\\n此階段所需時間  \\n\\\\[\\nt_2=\\\\frac{\\\\tfrac23}{0.96R}\\\\quad(\\\\text{天}) .\\n\\\\]\\n\\n已知總時間為 185 天，所以  \\n\\\\[\\nt_1+t_2 = \\\\frac{1}{3R}+\\\\frac{\\\\tfrac23}{0.96R}=185 .\\n\\\\]\\n\\n化簡分子：  \\n\\\\[\\n\\\\frac13+\\\\frac{\\\\tfrac23}{0.96}\\n=\\\\frac13+\\\\frac{\\\\tfrac23\\\\times\\\\frac{25}{24}}\\n=\\\\frac13+\\\\frac{25}{36}\\n=\\\\frac{12}{36}+\\\\frac{25}{36}\\n=\\\\frac{37}{36}.\\n\\\\]\\n\\n因此  \\n\\\\[\\n\\\\frac{37}{36}\\\\cdot\\\\frac{1}{R}=185\\n\\\\;\\\\Longrightarrow\\\\;\\nR=\\\\frac{37}{36\\\\times185}=\\\\frac{1}{180}\\\\quad(\\\\text{隧道/天}) .\\n\\\\]\\n\\n原始總時間  \\n\\\\[\\nT=\\\\frac{1}{R}=180\\\\quad(\\\\text{天}) .\\n\\\\]\\n\\n驗算：  \\n- 第一階段時間：\\\\(\\\\frac{1}{3}\\\\div\\\\frac{1}{180}=60\\\\) 天。  \\n- 第二階段時間：\\\\(\\\\frac{2}{3}\\\\div0.96\\\\cdot\\\\frac{1}{180}=125\\\\) 天。  \\n兩階段相加 \\\\(60+125=185\\\\) 天，符合已知總時間。\\n\\n**答案**  \\n若不使用新設備，整條隧道將會在 180 天內完成。  \\n\\\\boxed{180\\\\text{ 天}}</think>\\n\\n讓 \\\\(T\\\\) 為完成本隧道所需的原始日數。原始施工速率為 \\\\( \\\\frac{1}{T} \\\\) 隧道/日。當本隧道完成三分之一時，使用原始速率所耗時間為：\\\\(\\\\frac{T}{3}\\\\) 日。\\n\\n剩餘三分之二的隧道，施工速率提高了 20%，每日工作時間縮減至原來的 80%。新的施工速率為 \\\\(1.2 \\\\times \\\\frac{1}{T}\\\\) 隧道/小時，但由於工作時間減少，實際有效每日施工進度為：\\\\(1.2 \\\\times \\\\frac{1}{T} \\\\times 0.8 = \\\\frac{0.96}{T}\\\\) 隧道/日。\\n\\n按照新的速率完成剩餘三分之二的隧道所需時間為：\\\\(\\\\frac{2/3}{0.96/T} = \\\\frac{2T}{3 \\\\times 0.96} = \\\\frac{2T}{2.88} = \\\\frac{25T}{36}\\\\) 日。\\n\\n總用 185 日完成本隧道的時間為：\\\\(\\\\frac{T}{3} + \\\\frac{25T}{36} = 185\\\\)。\\n\\n將項合併：\\\\(\\\\frac{12T}{36} + \\\\frac{25T}{36} = \\\\frac{37T}{36} = 185\\\\)。\\n\\n求得 \\\\(T\\\\)：\\\\(T = 185 \\\\times \\\\frac{36}{37} = 180\\\\)。\\n\\n則若未使用新設備並沿原始速率施工，完成本隧道將需要 \\\\(\\\\boxed{180}\\\\) 日。\"\n",
+      "  }\n",
+      "]\n"
+     ]
+    }
+   ],
+   "source": [
+    "# Print the translated messages for the first example in formatted JSON\n",
+    "print(json.dumps(df['messages_zh'].iloc[0], ensure_ascii=False, indent=2))"
+   ]
+  }
+ ],
+ "metadata": {
+  "kernelspec": {
+   "display_name": "curator",
+   "language": "python",
+   "name": "python3"
+  },
+  "language_info": {
+   "codemirror_mode": {
+    "name": "ipython",
+    "version": 3
+   },
+   "file_extension": ".py",
+   "mimetype": "text/x-python",
+   "name": "python",
+   "nbconvert_exporter": "python",
+   "pygments_lexer": "ipython3",
+   "version": "3.12.11"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 5
+}
diff --git a/examples/translation.py b/examples/translation.py
new file mode 100644
index 000000000..01f9fe7e3
--- /dev/null
+++ b/examples/translation.py
@@ -0,0 +1,48 @@
+from nemo_curator.synthetic.translate import TranslationDataGenerator
+
+if __name__ == "__main__":
+    # Function-based usage
+    text = "Once upon a time, there were three little pig brothers..."
+    generator = TranslationDataGenerator(
+        base_url="http://localhost:11434/v1",
+        api_key="",
+        init_translate_model="gpt-oss:latest",
+        reflection_model="gpt-oss:latest",
+        improvement_model="gpt-oss:latest",
+        hf_tokenizer="openai/gpt-oss-20b",
+        hf_token="",
+        temperature=1.0,
+        top_p=1.0,
+        max_tokens=24576,
+        max_token_per_chunk=5000,
+        source_lang="English",
+        target_lang="Traditional Chinese",
+        country="Taiwan",
+    )
+    translations = generator.generate(text)
+    print(generator.parse_response(translations))
+
+    # YAML-based usage (parameters only, text provided in code)
+    generator_yaml = TranslationDataGenerator.from_yaml("config/translation_config.yaml")
+    print(generator_yaml.generate_from_yaml("config/translation_config.yaml", text))
+
+    # Pipeline DataFrame usage
+    import pandas as pd
+    df = pd.DataFrame(
+        {
+            "text": [
+                "Once upon a time, there were three little pig brothers...",
+                "The quick brown fox jumps over the lazy dog.",
+            ]
+        }
+    )
+    df_translated = generator_yaml.generate_from_dataframe(df, text_column="text", batch_size=16)
+    print(df_translated.head())
+
+    # Async test for async_generate_from_dataframe
+    import asyncio
+    async def test_async_generate_from_dataframe():
+        df_translated = await generator_yaml.async_generate_from_dataframe(df, text_column="text", batch_size=16)
+        print("[Async]\n", df_translated.head())
+
+    asyncio.run(test_async_generate_from_dataframe())
\ No newline at end of file
diff --git a/nemo_curator/synthetic/prompts.py b/nemo_curator/synthetic/prompts.py
index b4092a50c..38b34b3f6 100644
--- a/nemo_curator/synthetic/prompts.py
+++ b/nemo_curator/synthetic/prompts.py
@@ -137,3 +137,82 @@
 
 Task:
 Extract the factual information, concrete details, and key concepts from the above text following the instructions."""
+
+INITIAL_TRANSLATION_PROMPT = """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}:"""
+
+REFLECTION_COUNTRY_TRANSLATION_PROMPT = """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. \
+The final style and tone of the translation should match the style of {target_lang} colloquially spoken in {country}.
+
+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 take 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.\nEach suggestion should address one specific part of the translation.\nOutput only the suggestions and nothing else."""
+
+REFLECTION_TRANSLATION_PROMPT = """Your task is to carefully read a source text and a translation from {source_lang} to {target_lang}, and then give constructive criticisms and helpful suggestions to improve the translation. \
+
+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 take 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.\nEach suggestion should address one specific part of the translation.\nOutput only the suggestions and nothing else."""
+
+IMPROVE_TRANSLATION_PROMPT = """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.
+
+Please translate the text and output only in the following format, with no additional content, no keys, no quotes, and no JSON:
+<IMPROVED_TRANSLATION>
+...
+</IMPROVED_TRANSLATION>
+"""
\ No newline at end of file
diff --git a/nemo_curator/synthetic/translate.py b/nemo_curator/synthetic/translate.py
new file mode 100644
index 000000000..14949f5aa
--- /dev/null
+++ b/nemo_curator/synthetic/translate.py
@@ -0,0 +1,373 @@
+"""
+The work logic and prompts here are referenced from [Github: andrewyng/translation-agent](https://github.com/andrewyng/translation-agent).
+"""
+
+import yaml
+import re
+from nemo_curator.synthetic.generator import SyntheticDataGenerator
+from nemo_curator.synthetic.prompts import (
+    INITIAL_TRANSLATION_PROMPT,
+    REFLECTION_COUNTRY_TRANSLATION_PROMPT,
+    REFLECTION_TRANSLATION_PROMPT,
+    IMPROVE_TRANSLATION_PROMPT,
+)
+from nemo_curator import OpenAIClient
+from openai import OpenAI
+from transformers import AutoTokenizer
+import spacy
+
+def extract_content(s:str, debug:bool=False) -> str|None:
+    match = re.search(r'<IMPROVED_TRANSLATION>(.*?)</IMPROVED_TRANSLATION>', s, re.DOTALL)
+    if match:
+        extracted = match.group(1).strip()
+        if debug:
+            print(f"EXTRACTED: {extracted}")
+        return extracted
+    else:
+        if "</IMPROVED_TRANSLATION>" in s:
+            s = s.replace("</IMPROVED_TRANSLATION>", "")
+            if debug:
+                print(f"EXTRACTED (no closing tag): remove </IMPROVED_TRANSLATION>")
+        elif "<IMPROVED_TRANSLATION>" in s:
+            s = s.replace("<IMPROVED_TRANSLATION>", "")
+            if debug:
+                print(f"EXTRACTED (no opening tag): remove <IMPROVED_TRANSLATION>")
+        else:
+            if debug:
+                print("No tags found.")
+        return s
+
+class TextSplitter:
+    """
+    Utility class for splitting text into chunks based on token count.
+    """
+    def __init__(self, model_name: str = "", hf_token: str = "", max_token_per_chunk: int = 4096, tokenizer: AutoTokenizer = None, spacy_model: str = "xx_sent_ud_sm"):
+        self.model_name = model_name
+        self.hf_token = hf_token
+        self.max_token_per_chunk = max_token_per_chunk
+        self.nlp = spacy.load(spacy_model)
+        # Load tokenizer for counting tokens
+        if tokenizer is not None:
+            self.tokenizer = tokenizer
+        elif self.model_name:
+            self.tokenizer = AutoTokenizer.from_pretrained(self.model_name, token=self.hf_token, use_fast=True)
+        else:
+            raise ValueError("Model name is empty. Please provide a valid model name.")
+
+    def split_sentences(self, text: str) -> list[str]:
+        # split sentences by spaCy
+        doc = self.nlp(text)
+        return [sent.text for sent in doc.sents]
+
+    def num_tokens_in_string_hf(self, input_str: str) -> int:
+        # Count tokens in a string using the tokenizer
+        return len(self.tokenizer.encode(input_str))
+
+    def split_long_text(self, text: str, max_tokens: int = None) -> list[str]:
+        # Split long text into chunks based on max token count
+        if max_tokens is None:
+            max_tokens = self.max_token_per_chunk
+        sentences = self.split_sentences(text)
+        chunks = []
+        current_chunk = ''
+        for sentence in sentences:
+            # Separate sentences with spaces
+            test_chunk = current_chunk + " " + sentence
+            token_count = self.num_tokens_in_string_hf(test_chunk)
+            if token_count > max_tokens:
+                if current_chunk:
+                    chunks.append(current_chunk)
+                current_chunk = sentence
+            else:
+                current_chunk = test_chunk
+        if current_chunk:
+            chunks.append(current_chunk)
+        return chunks
+
+class TranslationDataGenerator(SyntheticDataGenerator):
+    """
+    Synthetic data generator for translation tasks.
+    Supports both function-based and YAML-based configuration.
+    """
+    def __init__(
+        self,
+        base_url: str = "https://integrate.api.nvidia.com/v1",
+        api_key: str = "",
+        init_translate_model: str = "openai/gpt-oss-20b",
+        reflection_model: str = "openai/gpt-oss-20b",
+        improvement_model: str = "openai/gpt-oss-20b",
+        hf_tokenizer: str = "openai/gpt-oss-20b",
+        hf_token: str = "",
+        max_token_per_chunk: int = 5000,
+        temperature: float = 1.0,
+        top_p: float = 1.0,
+        max_tokens: int = 24576,
+        stop: list[str] = None,
+        source_lang: str = "English",
+        target_lang: str = "Traditional Chinese",
+        country: str = "Taiwan",
+        spacy_model: str = "xx_sent_ud_sm"
+    ):
+        super().__init__()
+        # Initialize parameters for translation pipeline
+        self.base_url = base_url
+        self.api_key = api_key
+        self.init_translate_model = init_translate_model
+        self.reflection_model = reflection_model
+        self.improvement_model = improvement_model
+        self.hf_tokenizer = hf_tokenizer
+        self.hf_token = hf_token
+        self.max_token_per_chunk = max_token_per_chunk
+        self.temperature = temperature
+        self.top_p = top_p
+        self.max_tokens = max_tokens
+        self.source_lang = source_lang
+        self.target_lang = target_lang
+        self.country = country
+        self.spacy_model = spacy_model
+        if stop is not None:
+            self.stop = stop
+            self.text_splitter = TextSplitter(model_name=self.hf_tokenizer, hf_token=self.hf_token, max_token_per_chunk=self.max_token_per_chunk, spacy_model=self.spacy_model)
+        else:
+            self.tokenizer = AutoTokenizer.from_pretrained(self.hf_tokenizer, use_auth_token=self.hf_token)
+            self.stop = [self.tokenizer.decode([self.tokenizer.eos_token_id])]
+            self.text_splitter = TextSplitter(model_name=self.hf_tokenizer, hf_token=self.hf_token, max_token_per_chunk=self.max_token_per_chunk, tokenizer=self.tokenizer, spacy_model=self.spacy_model)
+        self.openai_client = OpenAI(base_url=self.base_url, api_key=self.api_key)
+        self.client = OpenAIClient(self.openai_client)
+        
+
+    @classmethod
+    def from_yaml(cls, yaml_path: str) -> "TranslationDataGenerator":
+        """
+        Create a TranslationDataGenerator instance from a YAML configuration file.
+        Args:
+            yaml_path: Path to the YAML configuration file.
+        Returns:
+            TranslationDataGenerator instance.
+        """
+        with open(yaml_path, "r") as f:
+            config = yaml.safe_load(f)
+        # Remove 'text' key if present, only pass valid constructor args
+        config = dict(config)  # Make a copy
+        config.pop('text', None)
+        return cls(**config)
+
+    def generate(self, llm_prompt: str | list[str], debug: bool = False) -> list[str]:
+        """
+        Main pipeline for translation generation.
+        Args:
+            llm_prompt: The input text to be translated (str or list of str).
+        Returns:
+            List of improved translations.
+        """
+        if isinstance(llm_prompt, str):
+            chunks = self.text_splitter.split_long_text(llm_prompt, max_tokens=self.max_token_per_chunk)
+        else:
+            chunks = llm_prompt
+        results = []
+        for chunk in chunks:
+            initial = self._run_init_translation(chunk)
+            reflection = self._run_reflection(chunk, initial)
+            improved = self._run_improve_translation(chunk, initial, reflection)
+            extract_data = extract_content(improved, debug=debug)
+            results.append(extract_data)
+        return results
+
+    def generate_from_yaml(self, yaml_path: str, text: str) -> str:
+        """
+        Run the translation pipeline using parameters from a YAML configuration file, with text provided in code.
+        Args:
+            yaml_path: Path to the YAML file containing parameters.
+            text: The text to translate (provided in code).
+        Returns:
+            Concatenated translation string.
+        """
+        with open(yaml_path, "r") as f:
+            config = yaml.safe_load(f)
+        # Only use parameters from YAML, ignore any text key
+        translations = self.generate(text)
+        return self.parse_response(translations)
+
+    def parse_response(self, llm_response: str | list[str]) -> str:
+        """
+        Parse the LLM response(s) into a single formatted string.
+        Args:
+            llm_response: List of improved translations.
+        Returns:
+            Concatenated translation string.
+        """
+        if isinstance(llm_response, list):
+            return "\n".join([x for x in llm_response if isinstance(x, str) and x is not None])
+        return llm_response
+
+    def _run_init_translation(self, text: str) -> str:
+        # Run initial translation step
+        prompt = INITIAL_TRANSLATION_PROMPT.format(
+            source_lang=self.source_lang,
+            source_text=text,
+            target_lang=self.target_lang,
+        )
+        responses = self.client.query_model(
+            model=self.init_translate_model,
+            messages=[{"role": "user", "content": prompt}],
+            stop=self.stop,
+            temperature=self.temperature,
+            top_p=self.top_p,
+            max_tokens=self.max_tokens,
+        )
+        return responses[0]
+
+    def _run_reflection(self, text: str, initial_translation_result: str) -> str:
+        # Run reflection step to improve translation
+        if self.country:
+            prompt = REFLECTION_COUNTRY_TRANSLATION_PROMPT.format(
+                source_lang=self.source_lang,
+                source_text=text,
+                target_lang=self.target_lang,
+                translation_1=initial_translation_result,
+                country=self.country,
+            )
+        else:
+            prompt = REFLECTION_TRANSLATION_PROMPT.format(
+                source_lang=self.source_lang,
+                source_text=text,
+                target_lang=self.target_lang,
+                translation_1=initial_translation_result,
+            )
+        responses = self.client.query_model(
+            model=self.reflection_model,
+            messages=[{"role": "user", "content": prompt}],
+            stop=self.stop,
+            temperature=self.temperature,
+            top_p=self.top_p,
+            max_tokens=self.max_tokens,
+        )
+        return responses[0]
+
+    def _run_improve_translation(self, text: str, initial_translation_result: str, reflection_result: str) -> str:
+        # Run improvement step to finalize translation
+        prompt = IMPROVE_TRANSLATION_PROMPT.format(
+            source_lang=self.source_lang,
+            source_text=text,
+            target_lang=self.target_lang,
+            translation_1=initial_translation_result,
+            reflection=reflection_result,
+        )
+        responses = self.client.query_model(
+            model=self.improvement_model,
+            messages=[{"role": "user", "content": prompt}],
+            stop=self.stop,
+            temperature=self.temperature,
+            top_p=self.top_p,
+            max_tokens=self.max_tokens,
+        )
+        return responses[0]
+
+    def generate_from_dataframe(self, df, text_column: str, batch_size: int = 32, output_column: str = "translated_text", **kwargs) -> "pd.DataFrame":
+        """
+        Pipeline: Translate a pandas DataFrame column in batches and return DataFrame with new column.
+        Args:
+            df: pandas DataFrame containing text data.
+            text_column: Name of the column to translate.
+            batch_size: Number of rows per batch.
+            output_column: Name of the column to store translations.
+            kwargs: Extra arguments for self.generate.
+        Returns:
+            DataFrame with new column containing translations.
+        """
+        import pandas as pd
+        results = []
+        n = len(df)
+        for start in range(0, n, batch_size):
+            end = min(start + batch_size, n)
+            batch = df.iloc[start:end]
+            texts = batch[text_column].tolist()
+            batch_translations = self.generate(texts, **kwargs)
+            results.extend(batch_translations)
+        df[output_column] = results
+        return df
+
+    async def async_generate_from_dataframe(self, df, text_column: str, batch_size: int = 32, output_column: str = "translated_text", **kwargs):
+        """
+        Asynchronous pipeline: Translate a pandas DataFrame column in batches using async requests and return DataFrame with new column.
+        Args:
+            df: pandas DataFrame containing text data.
+            text_column: Name of the column to translate.
+            batch_size: Number of rows per batch.
+            output_column: Name of the column to store translations.
+            kwargs: Extra arguments for self.generate.
+        Returns:
+            DataFrame with new column containing translations.
+        """
+        import pandas as pd
+        import asyncio
+
+        async def async_generate(texts, **kwargs):
+            loop = asyncio.get_event_loop()
+            return await loop.run_in_executor(None, self.generate, texts, **kwargs)
+
+        n = len(df)
+        tasks = []
+        for start in range(0, n, batch_size):
+            end = min(start + batch_size, n)
+            batch = df.iloc[start:end]
+            texts = batch[text_column].tolist()
+            tasks.append(async_generate(texts, **kwargs))
+
+        results = []
+        batch_translations_list = await asyncio.gather(*tasks)
+        for batch_translations in batch_translations_list:
+            results.extend(batch_translations)
+        df[output_column] = results
+        return df
+
+# Example usage for both function and YAML config
+if __name__ == "__main__":
+    # Function-based usage
+    text = "Once upon a time, there were three little pig brothers..."
+    generator = TranslationDataGenerator(
+        base_url="http://localhost:8080/v1",                   # (Change this) Base URL for local API (P.S: Ollama supports the OpenAI API format.)
+        api_key="",                                             # API key (empty if not required)
+        init_translate_model="/home/user/.cache/llama.cpp/unsloth_gpt-oss-20b-GGUF_gpt-oss-20b-F16.gguf",                  # Initial translation model
+        reflection_model="/home/user/.cache/llama.cpp/unsloth_gpt-oss-20b-GGUF_gpt-oss-20b-F16.gguft",                      # Reflection model for improvement
+        improvement_model="/home/user/.cache/llama.cpp/unsloth_gpt-oss-20b-GGUF_gpt-oss-20b-F16.gguf",                     # Model for translation improvement
+        hf_tokenizer="openai/gpt-oss-20b" ,                     # (Change this) HuggingFace model for tokenization
+        hf_token=None,                                            # (Change this) HuggingFace authentication token
+        temperature=1.0,                                        # Sampling temperature for generation
+        top_p=1.0,                                              # Nucleus sampling parameter
+        max_tokens=24576,                                        # Maximum tokens for input
+        stop=["<|return|>","<|endoftext|>", "<|call|>"],        # (Change this) Stop TOKEN sequences
+        max_token_per_chunk=5000,                               # Max tokens per chunk for translation
+        source_lang="English",                                  # Source language
+        target_lang="Traditional Chinese",                      # Target language
+        country="Taiwan",                                       # (Optional) Country context for translation
+        spacy_model="xx_sent_ud_sm"                            # (default) spaCy model for sentence splitting
+    )
+
+    translations = generator.generate(text)
+    print(generator.parse_response(translations))
+
+    # # YAML-based usage (parameters only, text provided in code)
+    # generator_yaml = TranslationDataGenerator.from_yaml("config/translation_config.yaml")
+    # print(generator_yaml.generate_from_yaml("config/translation_config.yaml", text))
+
+    # # Pipeline DataFrame usage
+    # import pandas as pd
+    # df = pd.DataFrame(
+    #     {
+    #         "text": [
+    #             "Once upon a time, there were three little pig brothers...",
+    #             "The quick brown fox jumps over the lazy dog.",
+    #         ]
+    #     }
+    # )
+    # df_translated = generator_yaml.generate_from_dataframe(df, text_column="text", batch_size=16)
+    # print(df_translated.head())
+
+    # # Async test for async_generate_from_dataframe
+    # import asyncio
+    # async def test_async_generate_from_dataframe():
+    #     df_translated = await generator_yaml.async_generate_from_dataframe(df, text_column="text", batch_size=16)
+    #     print("[Async]\n", df_translated.head())
+
+    # asyncio.run(test_async_generate_from_dataframe())
\ No newline at end of file
diff --git a/tests/synthetic/test_translate.py b/tests/synthetic/test_translate.py
new file mode 100644
index 000000000..7e17ff6c0
--- /dev/null
+++ b/tests/synthetic/test_translate.py
@@ -0,0 +1,28 @@
+import pytest
+
+def test_translate_function():
+    from nemo_curator.synthetic.translate import TranslationDataGenerator
+    from unittest.mock import patch
+    text = "Once upon a time, there were three little pig brothers..."
+    generator = TranslationDataGenerator(
+        base_url="http://localhost:11434/v1",
+        api_key="",
+        init_translate_model="gpt-oss:latest",
+        reflection_model="gpt-oss:latest",
+        improvement_model="gpt-oss:latest",
+        hf_tokenizer="openai/gpt-oss-20b",
+        hf_token=None,
+        temperature=1.0,
+        top_p=1.0,
+        max_tokens=24576,
+        max_token_per_chunk=5000,
+        source_lang="English",
+        target_lang="Traditional Chinese",
+        country="Taiwan",
+    )
+    with patch.object(generator, "generate", return_value=["mocked", "translated", "chunks"]) as mock_generate:
+        translation_chunks = generator.generate(text)
+        result = generator.parse_response(translation_chunks)
+        assert isinstance(result, str)
+        assert len(result) > 0
+        assert result == "mocked\ntranslated\nchunks"