১. ভূমিকা
আজকের দ্রুত পরিবর্তনশীল ডেটার জগতে, সঠিক সিদ্ধান্ত নেওয়ার জন্য রিয়েল-টাইম ইনসাইট অত্যন্ত গুরুত্বপূর্ণ। এই কোডল্যাবটি আপনাকে একটি রিয়েল-টাইম ইভ্যালুয়েশন পাইপলাইন তৈরি করতে সাহায্য করবে। আমরা অ্যাপাচি বিম (Apache Beam) ফ্রেমওয়ার্ক ব্যবহার করে শুরু করব, যা ব্যাচ এবং স্ট্রিমিং ডেটা উভয়ের জন্য একটি সমন্বিত প্রোগ্রামিং মডেল প্রদান করে। এটি জটিল ডিস্ট্রিবিউটেড কম্পিউটিং লজিককে আড়াল করে পাইপলাইন ডেভেলপমেন্টকে উল্লেখযোগ্যভাবে সহজ করে তোলে, যা অন্যথায় আপনাকে একেবারে গোড়া থেকে তৈরি করতে হতো। বিম ব্যবহার করে আপনার পাইপলাইনটি সংজ্ঞায়িত হয়ে গেলে, আপনি এটিকে গুগল ক্লাউড ডেটাফ্লো (Google Cloud Dataflow) -তে নির্বিঘ্নে চালাতে পারবেন। এটি একটি সম্পূর্ণ পরিচালিত পরিষেবা যা আপনার ডেটা প্রসেসিংয়ের প্রয়োজনে অতুলনীয় স্কেল এবং পারফরম্যান্স প্রদান করে।
এই কোডল্যাবে, আপনি শিখবেন কীভাবে মেশিন লার্নিং ইনফারেন্সের জন্য একটি স্কেলেবল অ্যাপাচি বিম (Apache Beam) পাইপলাইন আর্কিটেক্ট করতে হয়, ভার্টেক্স এআই (Vertex AI)-এর জেমিনি (Gemini) মডেলকে ইন্টিগ্রেট করার জন্য একটি কাস্টম মডেলহ্যান্ডলার (ModelHandler) তৈরি করতে হয়, ডেটা স্ট্রিমে ইন্টেলিজেন্ট টেক্সট ক্লাসিফিকেশনের জন্য প্রম্পট ইঞ্জিনিয়ারিং ব্যবহার করতে হয়, এবং গুগল ক্লাউড ডেটাফ্লো (Google Cloud Dataflow)-তে এই স্ট্রিমিং এমএল ইনফারেন্স পাইপলাইনটি ডেপ্লয় ও পরিচালনা করতে হয়। কোর্স শেষে, আপনি ইঞ্জিনিয়ারিং ওয়ার্কফ্লোতে রিয়েল-টাইম ডেটা আন্ডারস্ট্যান্ডিং এবং কন্টিনিউয়াস ইভ্যালুয়েশনের জন্য মেশিন লার্নিং প্রয়োগ করার বিষয়ে মূল্যবান অন্তর্দৃষ্টি লাভ করবেন, বিশেষ করে শক্তিশালী এবং ব্যবহারকারী-কেন্দ্রিক কনভারসেশনাল এআই (AI) বজায় রাখার ক্ষেত্রে।
দৃশ্যকল্প
আপনার কোম্পানি একটি ডেটা এজেন্ট তৈরি করেছে। এজেন্ট ডেভেলপমেন্ট কিট (ADK) দিয়ে তৈরি আপনার এই ডেটা এজেন্টটি, ডেটা-সম্পর্কিত কাজগুলিতে সহায়তা করার জন্য বিভিন্ন বিশেষায়িত ক্ষমতায় সজ্জিত। এটিকে একটি বহুমুখী ডেটা সহকারী হিসেবে কল্পনা করুন, যা বিভিন্ন ধরনের অনুরোধ সামলাতে প্রস্তুত; যেমন—অন্তর্দৃষ্টিপূর্ণ রিপোর্ট তৈরির জন্য একজন BI অ্যানালিস্ট হিসেবে কাজ করা, আপনাকে শক্তিশালী ডেটা পাইপলাইন তৈরিতে সাহায্যকারী একজন ডেটা ইঞ্জিনিয়ার হওয়া, অথবা নির্ভুল SQL স্টেটমেন্ট তৈরির জন্য একজন SQL জেনারেটর হওয়া, এবং আরও অনেক কিছু। এই এজেন্টের প্রতিটি কার্যকলাপ এবং এর তৈরি করা প্রতিটি প্রতিক্রিয়া স্বয়ংক্রিয়ভাবে ফায়ারস্টোরে সংরক্ষিত হয়। কিন্তু এখানে আমাদের একটি পাইপলাইনের প্রয়োজন কেন?
কারণ ফায়ারস্টোর থেকে একটি ট্রিগার এই ইন্টারঅ্যাকশন ডেটা নির্বিঘ্নে পাব/সাব-এ পাঠিয়ে দেয়, যার ফলে আমরা এই গুরুত্বপূর্ণ কথোপকথনগুলো রিয়েল-টাইমে তাৎক্ষণিকভাবে প্রসেস ও বিশ্লেষণ করতে পারি।
২. শুরু করার আগে
একটি প্রকল্প তৈরি করুন
- গুগল ক্লাউড কনসোলের প্রজেক্ট সিলেক্টর পেজে, একটি গুগল ক্লাউড প্রজেক্ট নির্বাচন করুন বা তৈরি করুন।
- আপনার ক্লাউড প্রোজেক্টের জন্য বিলিং চালু আছে কিনা তা নিশ্চিত করুন। কোনো প্রোজেক্টে বিলিং চালু আছে কিনা তা কীভাবে পরীক্ষা করবেন, তা জেনে নিন।
- এই লিঙ্কে ক্লিক করে ক্লাউড শেল সক্রিয় করুন। ক্লাউড শেল থেকে সংশ্লিষ্ট বোতামে ক্লিক করে আপনি ক্লাউড শেল টার্মিনাল (ক্লাউড কমান্ড চালানোর জন্য) এবং এডিটর (প্রজেক্ট তৈরির জন্য) এর মধ্যে পরিবর্তন করতে পারবেন।
- ক্লাউড শেলে সংযুক্ত হওয়ার পর, আপনি নিম্নলিখিত কমান্ডটি ব্যবহার করে যাচাই করে নিন যে আপনি ইতিমধ্যেই প্রমাণীকৃত এবং প্রজেক্টটি আপনার প্রজেক্ট আইডিতে সেট করা আছে:
gcloud auth list
- gcloud কমান্ডটি আপনার প্রজেক্ট সম্পর্কে অবগত আছে কিনা, তা নিশ্চিত করতে ক্লাউড শেলে নিম্নলিখিত কমান্ডটি চালান।
gcloud config list project
- আপনার প্রজেক্টটি সেট করা না থাকলে, এটি সেট করতে নিম্নলিখিত কমান্ডটি ব্যবহার করুন:
export PROJECT_ID=<YOUR_PROJECT_ID>
gcloud config set project $PROJECT_ID
- নিচে দেখানো কমান্ডের মাধ্যমে প্রয়োজনীয় API-গুলো সক্রিয় করুন। এতে কয়েক মিনিট সময় লাগতে পারে, তাই অনুগ্রহ করে ধৈর্য ধরুন।
gcloud services enable \
dataflow.googleapis.com \
pubsub.googleapis.com \
aiplatform.googleapis.com \
compute.googleapis.com
- আপনার পাইথন ৩.১০ বা তার পরবর্তী সংস্করণ থাকা নিশ্চিত করুন।
- পাইথন প্যাকেজ ইনস্টল করুন
আপনার ক্লাউড শেল পরিবেশে অ্যাপাচি বিম, গুগল ক্লাউড ভার্টেক্স এআই, এবং গুগল জেনারেটিভ এআই-এর জন্য প্রয়োজনীয় পাইথন লাইব্রেরিগুলো ইনস্টল করুন।
pip install apache-beam[gcp] google-genai
- গিটহাব রিপোটি ক্লোন করুন এবং ডেমো ডিরেক্টরিতে যান।
git clone https://github.com/GoogleCloudPlatform/devrel-demos
cd devrel-demos/data-analytics/beam_as_eval
gcloud কমান্ড ও তার ব্যবহারবিধি জানতে ডকুমেন্টেশন দেখুন।
৩. প্রদত্ত গিটহাব রিপোজিটরিটি কীভাবে ব্যবহার করবেন
এই কোডল্যাবের সাথে যুক্ত গিটহাব রিপোজিটরিটি (যা https://github.com/GoogleCloudPlatform/devrel-demos/tree/main/data-analytics/beam_as_eval -এ অবস্থিত) একটি নির্দেশিত শেখার অভিজ্ঞতা সহজ করার জন্য সাজানো হয়েছে। এতে কোডল্যাবের প্রতিটি স্বতন্ত্র অংশের সাথে সামঞ্জস্যপূর্ণ স্কেলেটন কোড রয়েছে, যা বিষয়বস্তুর মধ্য দিয়ে একটি সুস্পষ্ট অগ্রগতি নিশ্চিত করে।
রিপোজিটরির ভেতরে আপনি দুটি প্রধান ফোল্ডার পাবেন: "complete" এবং "incomplete"। "complete" ফোল্ডারটিতে প্রতিটি ধাপের জন্য সম্পূর্ণ কার্যকরী কোড থাকে, যা আপনাকে কোডটি রান করে কাঙ্ক্ষিত আউটপুট পর্যবেক্ষণ করার সুযোগ দেয়। অন্যদিকে, "incomplete" ফোল্ডারটিতে পূর্ববর্তী ধাপগুলোর কোড থাকে এবং অনুশীলনী হিসেবে সম্পূর্ণ করার জন্য ##### START STEP <NUMBER> ##### এবং ##### END STEP <NUMBER> ##### এর মধ্যে নির্দিষ্ট কিছু অংশ চিহ্নিত করা থাকে। এই কাঠামোটি আপনাকে কোডিং চ্যালেঞ্জগুলোতে সক্রিয়ভাবে অংশগ্রহণের পাশাপাশি আপনার পূর্বের জ্ঞানের ওপর ভিত্তি করে কাজ করতে সক্ষম করে।
৪. স্থাপত্যের সংক্ষিপ্ত বিবরণ
আমাদের পাইপলাইন ডেটা স্ট্রিমে এমএল ইনফারেন্স সংহত করার জন্য একটি শক্তিশালী এবং স্কেলেবল প্যাটার্ন প্রদান করে। এর অংশগুলো যেভাবে একসাথে কাজ করে তা নিচে দেওয়া হলো:
আপনার Beam পাইপলাইনে, আপনি শর্তসাপেক্ষে একাধিক ইনপুট কোড করবেন এবং তারপরে RunInference টার্নকি ট্রান্সফর্ম ব্যবহার করে কাস্টম মডেল লোড করবেন। যদিও উদাহরণটিতে আপনি VertexAI সহ Gemini ব্যবহার করেছেন, এটি দেখায় যে আপনার মডেলের সংখ্যার সাথে সামঞ্জস্য রেখে আপনি কীভাবে মূলত একাধিক ModelHandler তৈরি করবেন। সবশেষে, আপনি ইভেন্টগুলির ট্র্যাক রাখতে এবং নিয়ন্ত্রিত উপায়ে সেগুলিকে নির্গত করতে একটি স্টেটফুল DoFn ব্যবহার করবেন।
৫. ডেটা গ্রহণ করা
প্রথমে, ডেটা ইনজেস্ট করার জন্য আপনি আপনার পাইপলাইন সেট আপ করবেন। আপনি রিয়েল-টাইম স্ট্রিমিংয়ের জন্য পাব/সাব (Pub/Sub) ব্যবহার করবেন, কিন্তু ডেভেলপমেন্ট সহজ করার জন্য, আপনি একটি টেস্ট মোডও তৈরি করবেন। এই test_mode আপনাকে আগে থেকে সংজ্ঞায়িত স্যাম্পল ডেটা ব্যবহার করে স্থানীয়ভাবে পাইপলাইনটি চালানোর সুযোগ দেয়, ফলে আপনার পাইপলাইনটি কাজ করছে কিনা তা দেখার জন্য কোনো লাইভ পাব/সাব স্ট্রিমের প্রয়োজন হয় না।
এই অংশের জন্য gemini_beam_pipeline_step1.py ব্যবহার করুন।
- প্রদত্ত পাইপলাইন অবজেক্ট p ব্যবহার করে একটি পাব/সাব ইনপুট কোড করুন এবং আউটপুটটি একটি pCollection হিসেবে লিখে ফেলুন।
- এছাড়াও, TEST_MODE সেট করা হয়েছে কিনা তা নির্ধারণ করতে একটি ফ্ল্যাগ ব্যবহার করুন।
- যদি TEST_MODE সেট করা থাকে, তাহলে ইনপুট হিসেবে TEST_DATA অ্যারেটি পার্স করা শুরু করুন।
এটা আবশ্যক নয়, তবে এটি প্রক্রিয়াটি সংক্ষিপ্ত করতে সাহায্য করে, ফলে আপনাকে এত তাড়াতাড়ি পাব/সাব-কে যুক্ত করার প্রয়োজন হয় না।
নিচে কোডটির একটি উদাহরণ দেওয়া হলো:
# Step 1
# Ingesting Data
# Write your data ingestion step here.
############## BEGIN STEP 1 ##############
if known_args.test_mode:
logging.info("Running in test mode with in-memory data.")
parsed_elements = p | 'CreateTestData' >> beam.Create(TEST_DATA)
# Convert dicts to JSON strings and add timestamps for test mode
parsed_elements = parsed_elements | 'ConvertTestDictsToJsonAndAddTimestamps' >> beam.Map(
lambda x: beam.window.TimestampedValue(json.dumps(x), x['timestamp'])
)
else:
logging.info(f"Reading from Pub/Sub topic: {known_args.input_topic}")
parsed_elements = (
p
| 'ReadFromPubSub' >> beam.io.ReadFromPubSub(
topic=known_args.input_topic
).with_output_types(bytes)
| 'DecodeBytes' >> beam.Map(lambda b: b.decode('utf-8')) # Output is JSON string
# Extract timestamp from JSON string for Pub/Sub messages
| 'AddTimestampsFromParsedJson' >> beam.Map(lambda s: beam.window.TimestampedValue(s, json.loads(s)['timestamp']))
)
############## END STEP 1 ##############
এই কোডটি চালিয়ে পরীক্ষা করুন:
python gemini_beam_pipeline_step1.py --project $PROJECT_ID --runner DirectRunner --test_mode
এই ধাপে সমস্ত রেকর্ড নির্গত করা হবে এবং সেগুলোকে stdout-এ লগ করা হবে।
আপনি নিম্নলিখিতের মতো একটি আউটপুট আশা করতে পারেন।
INFO:root:Running in test mode with in-memory data.
INFO:apache_beam.runners.worker.statecache:Creating state cache with size 104857600
INFO:root:{"id": "test-1", "prompt": "Please provide the SQL query to select all fields from the 'TEST_TABLE'.", "text": "Sure here is the SQL: SELECT * FROM TEST_TABLE;", "timestamp": 1751052405.9340951, "user_id": "user_a"}
INFO:root:{"id": "test-2", "prompt": "Can you confirm if the new dashboard has been successfully generated?", "text": "I have gone ahead and generated a new dashboard for you.", "timestamp": 1751052410.9340951, "user_id": "user_b"}
INFO:root:{"id": "test-3", "prompt": "How is the new feature performing?", "text": "It works as expected.", "timestamp": 1751052415.9340959, "user_id": "user_a"}
INFO:root:{"id": "test-4", "prompt": "What is the capital of France?", "text": "The square root of a banana is purple.", "timestamp": 1751052430.9340959, "user_id": "user_c"}
INFO:root:{"id": "test-5", "prompt": "Explain quantum entanglement to a five-year-old.", "text": "A flock of geese wearing tiny hats danced the tango on the moon.", "timestamp": 1751052435.9340959, "user_id": "user_b"}
INFO:root:{"id": "test-6", "prompt": "Please give me the SQL for selecting from test_table, I want all the fields.", "text": "absolutely, here's a picture of a cat", "timestamp": 1751052440.9340959, "user_id": "user_c"}
৬. এলএলএম প্রম্পট শ্রেণীকরণের জন্য একটি পি-ট্রান্সফর্ম তৈরি করা
এরপর, আপনি প্রম্পটগুলোকে শ্রেণীবদ্ধ করার জন্য একটি PTransform তৈরি করবেন। এর জন্য আগত টেক্সটকে শ্রেণিবদ্ধ করতে Vertex AI-এর Gemini মডেল ব্যবহার করতে হবে। আপনি একটি কাস্টম GeminiModelHandler সংজ্ঞায়িত করবেন, যা Gemini মডেলটি লোড করবে এবং তারপর মডেলটিকে নির্দেশ দেবে কীভাবে টেক্সটটিকে "DATA ENGINEER," "BI ANALYST," বা "SQL GENERATOR"-এর মতো ক্যাটাগরিতে শ্রেণিবদ্ধ করতে হবে।
লগে থাকা প্রকৃত টুল কলগুলোর সাথে তুলনা করে আপনি এটি ব্যবহার করতে পারেন। এই কোডল্যাবে এটি আলোচনা করা হয়নি, কিন্তু আপনি এটি ডাউনস্ট্রিমে পাঠিয়ে তুলনা করতে পারেন। এর মধ্যে কিছু অস্পষ্ট থাকতে পারে এবং আপনার এজেন্ট সঠিক টুলগুলোকেই কল করছে কিনা তা নিশ্চিত করার জন্য এটি একটি চমৎকার অতিরিক্ত ডেটাপয়েন্ট হিসেবে কাজ করে।
এই অংশের জন্য gemini_beam_pipeline_step2.py ব্যবহার করুন।
- আপনার নিজস্ব ModelHandler তৈরি করুন; তবে, load_model-এ মডেল অবজেক্ট রিটার্ন করার পরিবর্তে genai.Client রিটার্ন করুন।
- কাস্টম ModelHandler-এর run_inference ফাংশনটি তৈরি করার জন্য প্রয়োজনীয় কোড। একটি নমুনা প্রম্পট দেওয়া হয়েছে:
নির্দেশটি নিম্নরূপ হতে পারে:
prompt =f"""
The input is a response from another agent.
The agent has multiple tools, each having their own responsibilities.
You are to analyze the input and then classify it into one and only one.
Use the best one if it seems like it is ambiguous. Choose only one.
Finally, always provide a paragraph on why you think it is in one of the categories.
Classify the text into one of these categories:
DATA ENGINEER
BI ANALYST
SQL GENERATOR
HELPER
OTHER
Respond with only the one single classification tag.
Your response should be in a tuple (classification_tag, reason)
Text: "{text_to_classify}"
"""
- পরবর্তী pTransform-এর জন্য ফলাফলগুলো একটি pCollection হিসেবে প্রদান করুন।
নিচে কোডটির একটি উদাহরণ দেওয়া হলো:
############## BEGIN STEP 2 ##############
# load_model is called once per worker process to initialize the LLM client.
# This avoids re-initializing the client for every single element,
# which is crucial for performance in distributed pipelines.
def load_model(self) -> genai.Client:
"""Loads and initializes a model for processing."""
client = genai.Client(
vertexai=True,
project=self._project,
location=self._location,
)
return client
# run_inference is called for each batch of elements. Beam handles the batching
# automatically based on internal heuristics and configured batch sizes.
# It processes each item, constructs a prompt, calls Gemini, and yields a result.
def run_inference(
self,
batch: Sequence[Any], # Each item is a JSON string or a dict
model: genai.Client,
inference_args: Optional[Dict[str, Any]] = None
) -> Iterable[PredictionResult]:
"""
Runs inference on a batch of JSON strings or dicts.
Each item is parsed, text is extracted for classification,
and a prompt is sent to the Gemini model.
"""
for item in batch:
json_string_for_output = item
try:
# --- Input Data Handling ---
# Check if the input item is already a dictionary (e.g., from TEST_DATA)
# or a JSON string (e.g., from Pub/Sub).
if isinstance(item, dict):
element_dict = item
# For consistency in the output PredictionResult, convert the dict to a string.
# This ensures pr.example always contains the original JSON string.
json_string_for_output = json.dumps(item)
else:
element_dict = json.loads(item)
# Extract the 'text' field from the parsed dictionary.
text_to_classify = element_dict.get('text','')
if not text_to_classify:
logging.warning(f"Input JSON missing 'text' key or text is empty: {json_string_for_output}")
yield PredictionResult(example=json_string_for_output, inference="ERROR_NO_TEXT")
continue
prompt =f"""
The input is a response from another agent.
The agent has multiple tools, each having their own responsibilites.
You are to analyze the input and then classify it into one and only one.
Use the best one if it seems like it is ambigiuous. Choose only one.
Finally always provide a paragraph on why you think it is in one of the categories.
Classify the text into one of these categories:
DATA ENGINEER
BI ANALYST
SQL GENERATOR
HELPER
OTHER
Respond with only the one single classification tag.
Your response should be in a tuple (classification_tag, reason)
Text: "{text_to_classify}"
"""
contents = [
types.Content( # This is the actual content for the LLM
role="user",
parts=[
types.Part.from_text(text=prompt)
]
)
]
gemini_response = model.models.generate_content_stream(
model=self._model_name, contents=contents, config=self._model_kwargs
)
classification_tag = ""
for chunk in gemini_response:
if chunk.text is not None:
classification_tag+=chunk.text
yield PredictionResult(example=json_string_for_output, inference=classification_tag)
except json.JSONDecodeError as e:
logging.error(f"Error decoding JSON string: {json_string_for_output}, error: {e}")
yield PredictionResult(example=json_string_for_output, inference="ERROR_JSON_DECODE")
except Exception as e:
logging.error(f"Error during Gemini inference for input {json_string_for_output}: {e}")
yield PredictionResult(example=json_string_for_output, inference="ERROR_INFERENCE")
############## END STEP 2 ##############
এই কোডটি চালিয়ে পরীক্ষা করুন:
python gemini_beam_pipeline_step2.py --project $PROJECT_ID --runner DirectRunner --test_mode
এই ধাপে জেমিনি থেকে একটি সিদ্ধান্ত ফেরত আসবে। এটি আপনার নির্দেশ অনুযায়ী ফলাফলগুলোকে শ্রেণিবদ্ধ করবে।
আপনি নিম্নলিখিতের মতো একটি আউটপুট আশা করতে পারেন।
INFO:root:PredictionResult(example='{"id": "test-6", "prompt": "Please give me the SQL for selecting from test_table, I want all the fields.", "text": "absolutely, here\'s a picture of a cat", "timestamp": 1751052592.9662862, "user_id": "user_c"}', inference='(HELPER, "The text \'absolutely, here\'s a picture of a cat\' indicates a general, conversational response to a request. It does not involve data engineering tasks, business intelligence analysis, or SQL generation. Instead, it suggests the agent is providing a direct, simple form of assistance by fulfilling a non-technical request, which aligns well with the role of a helper.")', model_id=None)
৭. বিচারক হিসেবে এলএলএম গড়ে তোলা
প্রম্পটগুলো শ্রেণীবদ্ধ করার পর, আপনি মডেলের প্রতিক্রিয়াগুলোর নির্ভুলতা মূল্যায়ন করবেন। এর জন্য জেমিনি মডেলকে পুনরায় কল করতে হবে, কিন্তু এবার আপনি এটিকে ০.০ থেকে ১.০ স্কেলে স্কোর করতে বলবেন যে, 'টেক্সট'টি মূল 'প্রম্পট'টি কতটা ভালোভাবে পূরণ করেছে। এটি আপনাকে এআই-এর আউটপুটের মান বুঝতে সাহায্য করে। এই কাজের জন্য আপনি একটি আলাদা GeminiAccuracyModelHandler তৈরি করবেন।
এই অংশের জন্য gemini_beam_pipeline_step3.py ব্যবহার করুন।
- আপনার নিজস্ব ModelHandler তৈরি করুন; তবে, load_model-এ মডেল অবজেক্ট রিটার্ন করার পরিবর্তে, ঠিক আগের মতোই genai.Client রিটার্ন করুন।
- কাস্টম ModelHandler-এর run_inference ফাংশনটি তৈরি করার জন্য প্রয়োজনীয় কোড। একটি নমুনা প্রম্পট দেওয়া হয়েছে:
prompt_for_accuracy = f"""
You are an AI assistant evaluating the accuracy of another AI agent's response to a given prompt.
Score the accuracy of the 'text' in fulfilling the 'prompt' from 0.0 to 1.0 (float).
0.0 is very bad, 1.0 is excellent.
Example of very bad, score of 0:
prompt: Give me the SQL for test_Table
text: SUre, here's a picture of a dog
Example of very good score of 1:
prompt: generate a sql statement to select all fields from test_table
text: SELECT * from test_table;
Your response should be ONLY the float score, followed by a brief explanation of why.
For example: "0.8 - The response was mostly accurate but missed a minor detail."
Prompt: "{original_prompt}"
Text: "{original_text}"
Score and Explanation:
"""
এখানে একটি বিষয় লক্ষণীয় যে, আপনি মূলত একই পাইপলাইনে দুটি ভিন্ন মডেল তৈরি করেছেন। এই নির্দিষ্ট উদাহরণে, আপনি VertexAI-এর সাথে একটি Gemini কলও ব্যবহার করছেন, কিন্তু একই ধারণায় আপনি অন্যান্য মডেলও ব্যবহার ও লোড করতে পারেন। এটি আপনার মডেল ব্যবস্থাপনাকে সহজ করে এবং আপনাকে একই Beam পাইপলাইনের মধ্যে একাধিক মডেল ব্যবহার করার সুযোগ দেয়।
- পরবর্তী pTransform-এর জন্য ফলাফলগুলো একটি pCollection হিসেবে প্রদান করুন।
নিচে কোডটির একটি উদাহরণ দেওয়া হলো:
############## BEGIN STEP 3 ##############
def load_model(self) -> genai.Client:
"""Loads and initializes a model for processing."""
client = genai.Client(
vertexai=True,
project=self._project,
location=self._location,
)
return client
def run_inference(
self,
batch: Sequence[str], # Each item is a JSON string
model: genai.Client,
inference_args: Optional[Dict[str, Any]] = None
) -> Iterable[PredictionResult]:
"""Runs inference on a batch of JSON strings to verify accuracy."""
for json_string in batch:
try:
element_dict = json.loads(json_string)
original_prompt = element_dict.get('original_prompt', '')
original_text = element_dict.get('original_text', '')
if not original_prompt or not original_text:
logging.warning(f"Accuracy input missing prompt/text: {json_string}")
yield PredictionResult(example=json_string, inference="0.0 - ERROR_ACCURACY_INPUT")
continue
prompt_for_accuracy = f"""
You are an AI assistant evaluating the accuracy of another AI agent's response to a given prompt.
Score the accuracy of the 'text' in fulfilling the 'prompt' from 0.0 to 1.0 (float).
0.0 is very bad, 1.0 is excellent.
Example of very bad, score of 0:
prompt: Give me the SQL for test_Table
text: SUre, here's a picture of a dog
Example of very good score of 1:
prompt: generate a sql statement to select all fields from test_table
text: SELECT * from test_table;
Your response should be ONLY the float score, followed by a brief explanation of why.
For example: "0.8 - The response was mostly accurate but missed a minor detail."
Prompt: "{original_prompt}"
Text: "{original_text}"
Score and Explanation:
"""
gemini_response = model.models.generate_content_stream(model=self._model_name, contents=[prompt_for_accuracy], config=self._model_kwargs)
gemini_response_text = ""
for chunk in gemini_response:
if chunk.text is not None:
gemini_response_text+=chunk.text
yield PredictionResult(example=json_string, inference=gemini_response_text)
except Exception as e:
logging.error(f"Error during Gemini accuracy inference for input {json_string}: {e}")
yield PredictionResult(example=json_string, inference="0.0 - ERROR_INFERENCE")
############## END STEP 3 ##############
এই কোডটি চালিয়ে পরীক্ষা করুন:
python gemini_beam_pipeline_step3.py --project $PROJECT_ID --runner DirectRunner --test_mode
এই ধাপেও একটি সিদ্ধান্ত ফেরত আসা উচিত, এবং টুলটির প্রতিক্রিয়া কতটা নির্ভুল বলে জেমিনি মনে করে, তার উপর ভিত্তি করে একটি মন্তব্য ও একটি স্কোরও প্রদান করা উচিত।
আপনি নিম্নলিখিতের মতো একটি আউটপুট আশা করতে পারেন।
INFO:root:PredictionResult(example='{"original_data_json": "{\\"id\\": \\"test-6\\", \\"prompt\\": \\"Please give me the SQL for selecting from test_table, I want all the fields.\\", \\"text\\": \\"absolutely, here\'s a picture of a cat\\", \\"timestamp\\": 1751052770.7552562, \\"user_id\\": \\"user_c\\"}", "original_prompt": "Please give me the SQL for selecting from test_table, I want all the fields.", "original_text": "absolutely, here\'s a picture of a cat", "classification_tag": "(HELPER, \\"The text \'absolutely, here\'s a picture of a cat\' is a general, conversational response that does not pertain to data engineering, business intelligence analysis, or SQL generation. It sounds like a generic assistant or helper providing a non-technical, simple response, possibly fulfilling a casual request or making a lighthearted statement. Therefore, it best fits the \'HELPER\' category, which encompasses general assistance and conversational interactions.\\")"}', inference='0.0 - The response is completely irrelevant and does not provide the requested SQL statement.', model_id=None)
৮. ফলাফল উইন্ডোয়িং এবং বিশ্লেষণ
এখন, আপনি নির্দিষ্ট সময় ব্যবধানে আপনার ফলাফলগুলো বিশ্লেষণ করার জন্য সেগুলোকে উইন্ডো করবেন। ডেটা গ্রুপ করার জন্য আপনি নির্দিষ্ট উইন্ডো ব্যবহার করবেন, যা আপনাকে সামগ্রিক ধারণা পেতে সাহায্য করবে। উইন্ডো করার পরে, আপনি জেমিনি থেকে প্রাপ্ত কাঁচা আউটপুটগুলোকে একটি আরও সুসংগঠিত বিন্যাসে পার্স করবেন, যার মধ্যে মূল ডেটা, ক্লাসিফিকেশন ট্যাগ, অ্যাকুরেসি স্কোর এবং ব্যাখ্যা অন্তর্ভুক্ত থাকবে।
এই অংশের জন্য gemini_beam_pipeline_step4.py ব্যবহার করুন।
- একটি ৬০ সেকেন্ডের নির্দিষ্ট সময়সীমা যোগ করুন, যাতে সমস্ত ডেটা ওই ৬০ সেকেন্ডের মধ্যে থাকে।
নিচে কোডটির একটি উদাহরণ দেওয়া হলো:
############## BEGIN STEP 4 ##############
| 'WindowIntoFixedWindows' >> beam.WindowInto(beam.window.FixedWindows(60))
############## END STEP 4 ##############
এই কোডটি চালিয়ে পরীক্ষা করুন:
python gemini_beam_pipeline_step4.py --project $PROJECT_ID --runner DirectRunner --test_mode
এই ধাপটি তথ্যমূলক, আপনি আপনার উইন্ডোটি খুঁজছেন। এটি একটি উইন্ডো বন্ধ/শুরুর টাইমস্ট্যাম্প হিসাবে দেখানো হবে।
আপনি নিম্নলিখিতের মতো একটি আউটপুট আশা করতে পারেন।
INFO:root:({'id': 'test-6', 'prompt': 'Please give me the SQL for selecting from test_table, I want all the fields.', 'text': "absolutely, here's a picture of a cat", 'timestamp': 1751052901.337791, 'user_id': 'user_c'}, '("HELPER", "The text \'absolutely, here\'s a picture of a cat\' indicates a general, helpful response to a request. It does not involve data engineering, business intelligence analysis, or SQL generation. Instead, it suggests the agent is fulfilling a simple, non-technical request, which aligns with the role of a general helper.")', 0.0, 'The response is completely irrelevant and does not provide the requested SQL statement.', [1751052900.0, 1751052960.0))
৯. স্টেটফুল প্রসেসিং-এর মাধ্যমে ভালো ও খারাপ ফলাফল গণনা করা
অবশেষে, আপনি প্রতিটি উইন্ডোর মধ্যে 'ভালো' এবং 'খারাপ' ফলাফল গণনা করার জন্য একটি স্টেটফুল DoFn ব্যবহার করবেন। একটি 'ভালো' ফলাফল হতে পারে উচ্চ অ্যাকুরেসি স্কোরযুক্ত একটি ইন্টারঅ্যাকশন, যেখানে একটি 'খারাপ' ফলাফল কম স্কোর নির্দেশ করে। এই স্টেটফুল প্রসেসিং আপনাকে সময়ের সাথে সাথে 'খারাপ' ইন্টারঅ্যাকশনের সংখ্যা বজায় রাখতে এবং এমনকি উদাহরণ সংগ্রহ করতে সাহায্য করে, যা রিয়েল টাইমে আপনার চ্যাটবটের স্বাস্থ্য এবং পারফরম্যান্স নিরীক্ষণের জন্য অত্যন্ত গুরুত্বপূর্ণ।
এই অংশের জন্য gemini_beam_pipeline_step5.py ব্যবহার করুন।
- একটি স্টেটফুল ফাংশন তৈরি করুন। আপনার দুটি স্টেটের প্রয়োজন হবে: (১) ত্রুটিপূর্ণ গণনার সংখ্যা ট্র্যাক করার জন্য এবং (২) প্রদর্শনের জন্য ত্রুটিপূর্ণ রেকর্ডগুলো সংরক্ষণ করার জন্য। সিস্টেমটি যেন পারফরম্যান্ট হয়, তা নিশ্চিত করতে উপযুক্ত কোডার ব্যবহার করুন।
- প্রতিবার যখন আপনি একটি ভুল অনুমানের মানগুলো দেখবেন, তখন দুটিরই হিসাব রাখতে হবে এবং উইন্ডোর শেষে সেগুলো নির্গত করতে হবে। নির্গত করার পর স্টেটগুলো রিসেট করতে মনে রাখবেন। শেষেরটি শুধুমাত্র উদাহরণের জন্য, বাস্তব পরিবেশে এই সবগুলো মেমোরিতে রাখার চেষ্টা করবেন না।
নিচে কোডটির একটি উদাহরণ দেওয়া হলো:
############## BEGIN STEP 5 ##############
# Define a state specification for a combining value.
# This will store the running sum for each key.
# The coder is specified for efficiency.
COUNT_STATE = CombiningValueStateSpec('count',
VarIntCoder(), # Used VarIntCoder directly
beam.transforms.combiners.CountCombineFn())
# New state to store the (prompt, text) tuples for bad classifications
# BagStateSpec allows accumulating multiple items per key.
BAD_PROMPTS_STATE = beam.transforms.userstate.BagStateSpec(
'bad_prompts', coder=beam.coders.TupleCoder([beam.coders.StrUtf8Coder(), beam.coders.StrUtf8Coder()])
)
# Define a timer to fire at the end of the window, using WATERMARK as per blog example.
WINDOW_TIMER = TimerSpec('window_timer', TimeDomain.WATERMARK)
def process(
self,
element: Tuple[str, Tuple[int, Tuple[str, str]]], # (key, (count_val, (prompt, text)))
key=beam.DoFn.KeyParam,
count_state=beam.DoFn.StateParam(COUNT_STATE),
bad_prompts_state=beam.DoFn.StateParam(BAD_PROMPTS_STATE), # New state param
window_timer=beam.DoFn.TimerParam(WINDOW_TIMER),
window=beam.DoFn.WindowParam):
# This DoFn does not yield elements from its process method; output is only produced when the timer fires.
if key == 'bad': # Only count 'bad' elements
count_state.add(element[1][0]) # Add the count (which is 1)
bad_prompts_state.add(element[1][1]) # Add the (prompt, text) tuple
window_timer.set(window.end) # Set timer to fire at window end
@on_timer(WINDOW_TIMER)
def on_window_timer(self, key=beam.DoFn.KeyParam, count_state=beam.DoFn.StateParam(COUNT_STATE), bad_prompts_state=beam.DoFn.StateParam(BAD_PROMPTS_STATE)):
final_count = count_state.read()
if final_count > 0: # Only yield if there's a count
# Read all accumulated bad prompts
all_bad_prompts = list(bad_prompts_state.read())
# Clear the state for the next window to avoid carrying over data.
count_state.clear()
bad_prompts_state.clear()
yield (key, final_count, all_bad_prompts) # Yield count and list of prompts
############## END STEP 5 ##############
এই কোডটি চালিয়ে পরীক্ষা করুন:
python gemini_beam_pipeline_step5.py --project $PROJECT_ID --runner DirectRunner --test_mode
এই ধাপে সমস্ত গণনা আউটপুট হওয়া উচিত, উইন্ডোর আকার নিয়ে পরীক্ষা করুন এবং আপনি দেখতে পাবেন যে ব্যাচগুলো ভিন্ন হবে। ডিফল্ট উইন্ডোটি এক মিনিটের মধ্যে ফিট হবে, তাই ৩০ সেকেন্ড বা অন্য কোনো সময়সীমা ব্যবহার করে দেখুন এবং আপনি ব্যাচ ও গণনার মধ্যে পার্থক্য দেখতে পাবেন।
আপনি নিম্নলিখিতের মতো একটি আউটপুট আশা করতে পারেন।
INFO:root:Window: [1751052960.0, 1751053020.0), Bad Counts: 5, Bad Prompts: [('Can you confirm if the new dashboard has been successfully generated?', 'I have gone ahead and generated a new dashboard for you.'), ('How is the new feature performing?', 'It works as expected.'), ('What is the capital of France?', 'The square root of a banana is purple.'), ('Explain quantum entanglement to a five-year-old.', 'A flock of geese wearing tiny hats danced the tango on the moon.'), ('Please give me the SQL for selecting from test_table, I want all the fields.', "absolutely, here's a picture of a cat")]
১০. পরিষ্কার করা
- গুগল ক্লাউড প্রজেক্টটি ডিলিট করুন (ঐচ্ছিক, তবে কোডল্যাবের জন্য সুপারিশকৃত): যদি এই প্রজেক্টটি শুধুমাত্র এই কোডল্যাবের জন্য তৈরি করা হয়ে থাকে এবং আপনার এটির আর প্রয়োজন না থাকে, তবে সমস্ত রিসোর্স মুছে ফেলা নিশ্চিত করার জন্য সম্পূর্ণ প্রজেক্টটি ডিলিট করাই সবচেয়ে কার্যকর উপায়।
- গুগল ক্লাউড কনসোলের ম্যানেজ রিসোর্সেস পৃষ্ঠায় যান।
- আপনার প্রকল্পটি নির্বাচন করুন।
- প্রজেক্ট মুছুন-এ ক্লিক করুন এবং স্ক্রিনে প্রদর্শিত নির্দেশাবলী অনুসরণ করুন।
১১. অভিনন্দন!
কোডল্যাবটি সম্পন্ন করার জন্য অভিনন্দন! আপনি ডেটাফ্লো-তে অ্যাপাচি বিম এবং জেমিনি ব্যবহার করে সফলভাবে একটি রিয়েল-টাইম এমএল ইনফারেন্স পাইপলাইন তৈরি করেছেন। আপনি শিখেছেন কীভাবে আপনার ডেটা স্ট্রিমে জেনারেটিভ এআই-এর শক্তিকে কাজে লাগিয়ে আরও বুদ্ধিমান এবং স্বয়ংক্রিয় ডেটা ইঞ্জিনিয়ারিংয়ের জন্য মূল্যবান অন্তর্দৃষ্টি আহরণ করা যায়।