[TOC]

# Mở đầu
## Tài liệu

- https://lilianweng.github.io/posts/2023-06-23-agent/
- https://www.philschmid.de/
- CTFAgent: An LLM-powered Agent for CTF Challenge Solving
- https://www.promptingguide.ai/techniques/cot

...

## Setup và framework

- Langchain
- Google Gen AI SDK (vì đây là cái duy nhất free và mình éo có tiền mua của claude hay openAI)

### Hạ tầng và một số thứ liên quan
Bên DevOps VN đang kết hợp với DataOnline hỗ trợ giảm giá dịch vụ VPS và Server (ở thời điểm bài này được viết)

- https://devops.vn/devops-vietnam-x-dataonline-uu-dai-vps-gia-re-hang-dau/

# Langchain
Phần đầu của bài viết này sẽ nói về cơ bản về Langchain
## Chatbot cơ bản

Tạo một file `.env` trong thư mục hiện tại và lưu vào trong đó `GOOGLE_API_KEY`.

Khi ta gọi `ChatGoogleGenerativeAI` thì langchain sẽ tự động load api key vào.

from langchain_google_genai import ChatGoogleGenerativeAI 

llm = ChatGoogleGenerativeAI(model="gemini-2.5-flash")

respond = llm.invoke('The sky is? ')
print(respond.content) 


Một số keywords để truyền vào:

- temperature: temperature càng cao thì câu trả lời trả về "càng sáng tạo" hơn, nhưng cũng dễ sai hơn.
- max_tokens: giới hạn số tokens cho output

### Input/Output

Đầu vào và đầu ra trong hệ thống agents được chia làm 3 loại chính đó là `SystemMessage, HumanMessage, AIMessage`

`SystemMessage` bao gồm các chỉ dẫn hệ thống để Agent làm theo. Còn `HumanMessage` là prompt trực tiếp của người dùng, `AIMessage` là kết quả trả về của Agent.

`AIMessage` được dùng để lưu output trả về của model, điều này giúp cải thiện memory của agent.

Ví dụ:

from langchain_google_genai import ChatGoogleGenerativeAI 
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage

llm = ChatGoogleGenerativeAI(
    model="gemini-2.5-flash",
    temperature=0.5,
    max_tokens = 800
    )
msgs = []
msgs.append(HumanMessage(content = '2+2'))
res = llm.invoke(msgs)
print(res.content) 
msgs.append(AIMessage(content=res.content))
msgs.append(HumanMessage(content = '*3'))
res = llm.invoke(msgs)
print(res.content)

Chia làm 2 phase, đầu tiên nó nhận một đầu vào của user là `2+2`. Sau đó tính toán kết quả, kết quả này sẽ được thêm vào `msgs` để lưu sau đó ta tiếp tục prompt `*3`.

Nếu như không có `AIMessage` ở trước đó thì Agent sẽ không biết `*3` cho cái gì.

### Prompt Template

Để thiết kế prompt được đa dạng thì ta sẽ cần tới các template. Lợi ích khi sử dụng template đó là ta có thể truyền tham số vào đó để tạo nên các đoạn prompt theo ý muốn, ngoài ra nó cũng phục vụ mục đích tái sử dụng prompt.

from langchain_core.prompts import PromptTemplate

prompt = PromptTemplate.from_template(
    "Answer this: {question}"
)

print(prompt.format(question="2+2"))

Ngoài `PromptTemplate` ra ta còn có `ChatPromptTemplate`.

Nó tương tự như `PromptTemplate` nhưng dành cho chat messages.

Ví dụ:

from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are helpful"),
    ("human", "{question}")
])
messages = prompt.format_messages(
    question="2+2?"
)

[SystemMessage(content='You are helpful', additional_kwargs={}, response_metadata={}),
 HumanMessage(content='2+2?', additional_kwargs={}, response_metadata={})]

### Output format

Tại sao ta lại cần formatting output của agent?

- Để tái sử dụng và có sự thống nhất về mặt thiết kế.

Tái sử dụng ở đây có ý nghĩa khi ta build một multi-stage agent, tức là nó phải trải qua nhiều bước. Output của một agent này có thể trở thành đầu ra của agent kia hoặc đơn giản là ta cho agent loop lại các bước thực thi. Như vậy để việc giao tiếp giữa các agent hoặc agent giao tiếp với chính nó trở nên hiệu quả thì ta nên thống nhất một kiểu dữ liệu đầu vào/đầu ra chung.

Format như thế nào?

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_google_genai import ChatGoogleGenerativeAI

llm = ChatGoogleGenerativeAI(model="gemini-2.5-flash")

prompt = ChatPromptTemplate.from_messages([
    ("system", "Answer short"),
    ("human", "{question}")
])

chain = prompt | llm | StrOutputParser()

res = chain.invoke({"question": "2+2?"})

print(res)

Ép trả JSON:

prompt = ChatPromptTemplate.from_messages([
("system","Return JSON"),
("human", """
Question: {question}

Return:
{{
 "answer": number
}}
""")
])

Nâng cao hơn ta sẽ sử dụng Pydantic.

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser, PydanticOutputParser
from langchain_google_genai import ChatGoogleGenerativeAI
from pydantic import BaseModel, Field
from typing import List 
llm = ChatGoogleGenerativeAI(
    model="gemini-2.5-flash",
    temperature = 0.5)

class ToolCall(BaseModel): 
    tool: str = Field(description = "Tool name") 
    input: str = Field(description = "Tool input") 

struct_llm = llm.with_structured_output(ToolCall)

res = struct_llm.invoke("2**15") 
print(res.model_dump_json()) 

Xem thêm tại đây: [Pydantic - Viblo](https://viblo.asia/p/gioi-thieu-ve-pydantic-don-gian-hoa-viec-xac-thuc-du-lieu-trong-python-AZoJjrggJY7)

### Components chaining
Trong langchain có một khái niệm gọi là runnable. Mọi thành phần trong LangChain (Prompt, Model, Tool, Output Parser) đều là một Runnable. Tất cả chúng đều tuân thủ cùng một giao thức cho nên ta có thể chain lại bằng toán tử pipe `|`.

Các runnable có 3 kiểu phương thức thực thi là `invoke, batch` và `stream`. Ví dụ như sau:

- `invoke`: single input -> single output
- `batch` : list of inputs -> list of outputs.
- `stream`: single input -> Iterator(list of outputs)

Nói thêm: Có hai phương pháp lập trình phổ biến là Imperative và Declarative

Nói đơn giản thì: Imperative là phương pháp lập trình tập trung vào mô tả cách chương trình hoạt động hay còn gọi là lập trình mệnh lệnh. Còn Declarative là lập trình khai báo sẽ thể hiện logic của chương trình mà không mô tả cách chương trình hoạt động?

Xem bài viết sau: https://fireship.dev/imperative-vs-declarative-programming

Tác giả giải thích và cho một số ví dụ khá hay về hai kiểu lập trình này.

Như mọi người thấy thì Python có thể lập trình theo cả hai cách trên. Một số thì thiên về imperative hơn như C, còn HTML, SQL thì thiên về declarative hơn. Vậy tại sao mình lại nói về hai kiểu lập trình này?

--> Vì ta có thể áp dụng các kiểu tư duy trên vào phát triển một LLM Application.

Hai hướng thiết kế trong Langchain:


Imperative là như cách viết của ta nãy giờ, ta sẽ gọi từng bước

messages = prompt.format_messages(question="2+2")

res = llm.invoke(messages)

text = res.content

print(text)

Declarative là nối các bước lại:

chain = prompt | llm | StrOutputParser()

chain.invoke({"question": "2+2"})

## RAG
- https://github.com/langchain-ai/rag-from-scratch

RAG hay Retrieval-Augmented Generation (RAG) là một khả năng giúp năng cao khả năng của mô hình sinh bằng cách kết hợp với tri thức bên ngoài. Phương pháp này thực hiện bằng cách truy xuất thông tin liên quan từ kho tài liệu (tri thức) và sử dụng chúng cho quá trình sinh câu trả lời dựa trên LLMs.

Để thực hiện ý tưởng trên thì ta cần trải qua 2 công đoạn chính:

- Đầu tiên là indexing documents, ta cần đánh chỉ mục các tài liệu này một cách hợp lí làm sao cho con chatbot có thể tìm kiếm thông tin liên quan tới câu hỏi một cách nhanh chóng và chính xác.
- Tiếp theo là retrieval, truy xuất dữ liệu từ các chỉ mục và dùng làm context cho LLM để giải quyết câu hỏi.

Chi tiết từng phần như sau.

- Đầu tiên là bước extract, ta cần trích xuất thông tin từ dữ liệu thô.
- Dữ liệu thô này sau đó sẽ được chia nhỏ thành các chunks.
- Chuyển đổi các chunks này thành các vector embeddings.
- Lưu trữ các vector này vào trong một vector database.

Pipeline có thể hình dung đơn giản như sau:

- Tạo vector database: convert dữ liệu tri thức của ta thành các vector và lưu trữ chúng vào một vector database
- User input: user cung cấp câu truy vấn (query) cho chatbot
- Information retrieval: dựa trên truy vấn của user, hệ thống sẽ tiến hành quét qua các vector trong database để xác định các phân đoạn tri thức nào có ngữ nghĩa tương đồng với câu truy vấn của người dùng
- Combining data: kết hợp các phân đoạn tri thức đã được truy xuất với câu truy vấn của người dùng để tạo thành một prompt hoàn chỉnh
- Generate text: Câu prompt này sau đó sẽ được gửi đến LLM để tạo ra câu trả lời dựa trên thông tin đã được truy xuất và câu hỏi của người dùng.


### Thực hành

Mình sẽ thử triển khai một RAG Pipeline đơn giản.

Trước tiên cần cài đặt một môi trường ảo, ở đây mình sẽ dùng `conda`. Ngoài ra mọi người có thể thử sử dụng `uv`.

conda create -n ctf-agent python=3.12 -y
conda activate ctf-agent
pip install langchain langchain-google-genai langchain-community langchain-core
pip install langchain-text-splitters 

Tiếp theo kiểm tra mô hình LLM nào mà api key có thể truy cập được. Ở đây mình xài gemini api key (vì openai éo cho xài free mà phải nạp)

from google import genai 
from dotenv import load_dotenv
load_dotenv() 
import os
# CHECK IF API KEY IS OKAY OR NOT 
gemini_key = os.environ["GOOGLE_API_KEY"]

def check_gemini_model(api_key):
    client = genai.Client(api_key=api_key)
    usable = []
    try: 
        models = client.models.list() 
        for model in models: 
            if "gemini" not in model.name.lower(): 
                continue 
            try: 
                client.models.generate_content( 
                    model = model.name, 
                    contents = 'hi'
                )
                usable.append(model.name) 
            except Exception as e: 
                pass
    except Exception as e: 
        pass 
    return usable 

# gemini_use = check_gemini_model(gemini_key) 
# print(gemini_use) 
# ['models/gemini-2.5-flash', 'models/gemini-flash-latest', 'models/gemini-flash-lite-latest', 'models/gemini-2.5-flash-lite', 'models/gemini-3-flash-preview', 'models/gemini-3.1-flash-lite-preview', 'models/gemini-robotics-er-1.5-preview']
print("gemini ok")

Để xử lí dữ liệu tri thức, mình cần chuyển nó về dưới dạng văn bản thuần túy.

Có 2 lựa chọn cho kiểu văn bản đó là Markdown hoặc txt. Ở đây mình sẽ lựa chọn Markdown vì nó có cấu trúc tiêu đề rõ ràng hơn so với txt là một khối dữ liệu liền mạch.

Ví dụ một cấu trúc cho file markdown như sau:

# [Tên Challenge]

**Category:** [Web / Pwn / Rev / Crypto / Forensics / Misc] 

## Description
> [Mô tả của challenge do ban tổ chức cung cấp]

## Overview
[Tóm tắt ngắn gọn nhất về mục tiêu của bài và công nghệ/ngôn ngữ được sử dụng]

## Technical details
[Đi thẳng vào vấn đề kỹ thuật cốt lõi. BỎ QUA các bước recon cơ bản (như file, checksec, nmap, v.v.) trừ khi nó là trọng tâm của bài. Xác định ngay lỗ hổng/thuật toán chính là gì và giải thích ngắn gọn cơ chế lỗi. Ví dụ: "Chương trình bị lỗi Buffer Overflow tại hàm X do hàm Y không kiểm tra độ dài", hoặc "Hệ mã RSA sử dụng e quá nhỏ dẫn đến Broadcast Attack".]

## Proof-of-Concept
[Tuyệt đối không dump toàn bộ script exploit thành một khối duy nhất mà không giải thích. Bạn phải miêu tả từng bước khai thác và trích xuất đoạn code xử lý việc đó ra ngay bên dưới lời giải thích. 
Ví dụ:
- Bước 1: Leak địa chỉ libc thông qua lỗi Format String.
`[Đoạn code snippet thực hiện việc leak]`
- Bước 2: Tính toán base address và gọi system('/bin/sh').
`[Đoạn code snippet tương ứng]`]

## P/S
[Phần này KHÔNG bắt buộc. Chỉ thêm vào nếu có các Takeaways quan trọng, bài học rút ra, hoặc các notes về những cách giải thay thế (unintended solution).]

Dùng phương thức `TextLoader` và `DirectoryLoader` để load dữ liệu từ các file markdown này:

from langchain_community.document_loaders import TextLoader, DirectoryLoader
loader = DirectoryLoader(
    "./Documents",
    glob = "**/*.md",
    loader_cls = TextLoader,
    show_progress=True,
    use_multithreading=True,
    silent_errors=True
    )

documents = loader.load() 
print(f"Loaded {len(documents)} documents") 
print(documents[0].metadata) 
print(documents[0].page_content[:500]) 

Tiếp theo ta cần chia nhỏ các tài liệu này thành các chunks. Đối với file md ta sẽ sử dụng hai phương thức là `MarkdownHeaderTextSplitter` và `RecursiveCharacterTextSplitter`.

Do markdown chia section theo các header nên cần phải chunk bằng `MarkdownHeaderTextSplitter` trước để giữ nguyên cấu trúc của tài liệu. Sau đó mới tiếp tục chia nhỏ hơn bằng `RecursiveCharacterTextSplitter` để đảm bảo độ dài của mỗi chunk không vượt quá giới hạn cho phép của mô hình LLM.

from langchain_text_splitters import MarkdownHeaderTextSplitter, RecursiveCharacterTextSplitter
header_split = [ 
    ('#', 'h1'), 
    ('##', 'h2')
]

md_splitter = MarkdownHeaderTextSplitter( 
    headers_to_split_on=header_split 
)

sections = []

for doc in documents:
    splits = md_splitter.split_text(doc.page_content)
    for s in splits:
        s.metadata.update(doc.metadata)
        sections.append(s)

chunk_splitter = RecursiveCharacterTextSplitter(
    chunk_size=800, 
    chunk_overlap=150,
    separators = ["\n\n", "\n", ". ", " ", ""])

final_chunk = chunk_splitter.split_documents(sections)
print(final_chunk[0].metadata)

Sau khi chunk ta sẽ embed các chunks này thành các embedding vector và lưu chúng trong một vector database. Gemini có cung cấp một embedding model free là `gemini-embedding-001`.

Check model cho embedding vector:

import os
from dotenv import load_dotenv
from google import genai

load_dotenv()
client = genai.Client(api_key=os.environ["GOOGLE_API_KEY"])

model_name = "gemini-embedding-001"

try:
    result = client.models.embed_content(
        model=model_name,
        contents="RSA small exponent attack"
    )
    emb = result.embeddings[0].values
    print("OK:", model_name)
    print("Embedding length:", len(emb))
except Exception as e:
    print("FAILED:", model_name)
    print(type(e).__name__, e)

Google api key thì cho xài free `gemini-embedding-001` với embedding length là 3072.

Thực hiện các bước embed -> store -> search. Để phục vụ việc demo đơn giản thì mình sẽ sử dụng `Chroma` làm vector database. Trong Langchain có hỗ trợ mongodb thông qua phương thức `MongoDBAtlasVectorSearch` mà ta sẽ tìm hiểu ở phần sau của bài viết.

embeddings = GoogleGenerativeAIEmbeddings(
    model="models/gemini-embedding-001"
)
vector_store = Chroma.from_documents(
    documents = final_chunk, 
    embedding = embeddings,
    persist_directory="./chroma_db",
)

print("Done embedding and storing documents in Chroma.")

result = vector_store.similarity_search(
    "How do we recover phi(n) from d",
    k=3
)

for i, doc in enumerate(result):
    print("=" * 80)
    print(f"Result {i+1}")
    print(doc.metadata)
    print(doc.page_content[:500])

Full code demo:

from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_core.prompts import PromptTemplate
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_chroma import Chroma
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
from langchain_community.document_loaders import TextLoader, DirectoryLoader
from langchain_text_splitters import MarkdownHeaderTextSplitter, RecursiveCharacterTextSplitter
from langchain_google_genai import GoogleGenerativeAIEmbeddings
from pprint import pprint 


'''
các bước để build một hệ thống RAG như sau: 
Các bước build một hệ thống RAG:
1. Load dữ liệu
2. Chunking
3. Embedding + vector DB
4. Retrieval 
5. Prompt + LLM
'''


# Bước 1
loader = DirectoryLoader(
    "./Documents",
    glob = "**/*.md",
    loader_cls = TextLoader,
    show_progress=True,
    use_multithreading=True,
    silent_errors=True
    )

documents = loader.load() 
# print(f"Loaded {len(documents)} documents") 
# print(documents[0].metadata) 
# print(documents[0].page_content[:500]) 


# Bước 2
header_split = [ 
    ('#', 'h1'), 
    ('##', 'h2')
]

md_splitter = MarkdownHeaderTextSplitter( 
    headers_to_split_on=header_split 
)

sections = []

for doc in documents:
    splits = md_splitter.split_text(doc.page_content)
    for s in splits:
        s.metadata.update(doc.metadata)
        sections.append(s)



chunk_splitter = RecursiveCharacterTextSplitter(
    chunk_size=800, 
    chunk_overlap=150,
    separators = ["\n\n", "\n", ". ", " ", ""])

final_chunk = chunk_splitter.split_documents(sections)


# Bước 3+4
embeddings = GoogleGenerativeAIEmbeddings(
    model="models/gemini-embedding-001"
)
vector_store = Chroma.from_documents(
    documents = final_chunk, 
    embedding = embeddings,
    persist_directory="./chroma_db",
)

print("Done embedding and storing documents in Chroma.")

result = vector_store.similarity_search(
    "How do we recover phi(n) from d",
    k=3
)



retriever = vector_store.as_retriever(
    search_type="mmr",
    search_kwargs={"k": 3, "fetch_k": 10}
)


# Bước 5

def format_docs(docs):
    parts = []
    for i, doc in enumerate(docs, 1):
        source = doc.metadata.get("source", "unknown")
        h1 = doc.metadata.get("h1", "")
        h2 = doc.metadata.get("h2", "")
        h3 = doc.metadata.get("h3", "")

        header_path = " > ".join([x for x in [h1, h2, h3] if x])

        parts.append(
            f"[Document {i}]\n"
            f"Source: {source}\n"
            f"Section: {header_path}\n"
            f"Content:\n{doc.page_content}"
        )
    return "\n\n" + ("\n\n" + "-" * 80 + "\n\n").join(parts)


llm = ChatGoogleGenerativeAI(
    model = "models/gemini-2.5-flash",
    temperature = 0,
)


template = """
You are a helpful assistant for technical and Crypto CTF knowledge-base QA.

Answer the user's question using ONLY the provided context.
If the answer is not in the context, say clearly that you do not know based on the retrieved documents.

When possible:
- explain briefly but clearly
- mention the relevant section/source
- do not invent facts outside the context

Context:
{context}

Question:
{question}

Answer:
"""

prompt = PromptTemplate.from_template(template)

rag_chain = (
    {"context": retriever | format_docs,
    "question": RunnablePassthrough(),
    }
    | prompt 
    | llm 
    | StrOutputParser()
)

while True: 
    query = input("Enter your question (or 'exit' to quit): ")
    if query.lower() == "exit":
        break
    answer = rag_chain.invoke(query) 
    print("\nAnswer:\n", answer)

Ở trên ta dồn hết mọi thứ vào 1 file nhằm mục đích demo nhưng trên thực tế ta cần tách ra làm 2 chương trình độc lập với nhau.

Một bên phục vụ cho việc indexing và một bên phục vụ cho truy vấn. Nếu dồn vào như vậy thì mỗi lần chạy chương trình ta lại phải thực hiện bước index thêm một lần nữa gây lãng phí token.

## Advanced RAG

Keyword: Multi-query , RAG-Fusion

## LangGraph for memory

Chatbot cần memory? Giả định một tình huống như sau:

User: giải thích về rsa small root attack cho tôi 
Bot: ...
User: vậy điều kiện để attack là gì? 

Nếu bot không có memory thì nó sẽ hiểu câu trên như thế nào? Nó không biết attack mà user đề cập đến là gì để cho câu trả lời

--> LLM là stateless tức là mỗi lần prompt xong thì nó quên hết những gì vừa hỏi.

Vậy thiết kế và cài đặt như thế nào?

from typing import TypedDict, Annotated 
from langgraph.graph.message import add_messages
from langgraph.graph import StateGraph, START, END 
from langchain_core.messages import HumanMessage, AIMessage 

from langchain_google_genai import ChatGoogleGenerativeAI

llm = ChatGoogleGenerativeAI(
    model="gemini-2.5-flash",
    temperature=0,
)


class State(TypedDict): 
    messages: Annotated[list, add_messages] 

builder = StateGraph(State) 

def chatbot(state: State):
    answer = llm.invoke(state["messages"])
    return {"messages": answer}


builder.add_node("chatbot",chatbot)
builder.add_edge(START, "chatbot")
builder.add_edge("chatbot", END)
graph = builder.compile() 

png_bytes = graph.get_graph().draw_mermaid_png()

with open("graph.png", "wb") as f:
    f.write(png_bytes)

Stateful memory :

from typing import TypedDict, Annotated 
from langgraph.graph.message import add_messages
from langgraph.graph import StateGraph, START, END 
from langchain_core.messages import HumanMessage, AIMessage 
from langgraph.checkpoint.memory import MemorySaver
from langchain_google_genai import ChatGoogleGenerativeAI

llm = ChatGoogleGenerativeAI(
    model="gemini-2.5-flash",
    temperature=0,
)


class State(TypedDict): 
    messages: Annotated[list, add_messages] 

builder = StateGraph(State) 

def chatbot(state: State):
    answer = llm.invoke(state["messages"])
    return {"messages": answer}


builder.add_node("chatbot",chatbot)
builder.add_edge(START, "chatbot")
builder.add_edge("chatbot", END)
graph = builder.compile(checkpointer=MemorySaver())

thread1  = {"configurable": {"thread_id": "1"}}
result_1 = graph.invoke(
    {"messages": [HumanMessage('hi, my name is Duc')]}, 
    config = thread1 
)

result_2 = graph.invoke(
    {"messages": [HumanMessage('what is my name?')]}, 
    config = thread1 
)
print(result_1["messages"][-1].content) 
print(result_2["messages"][-1].content)
png_bytes = graph.get_graph().draw_mermaid_png()
with open("graph.png", "wb") as f:
    f.write(png_bytes)