ai
LangChain Agent Automation Script — วิธีสร้าง AI
LangChain Agent
LangChain Agent ใช้ LLM ตัดสินใจเลือก Tools และ Actions อัตโนมัติ วิเคราะห์คำถาม เลือก Tool เรียกใช้ ดูผลลัพธ์ ตัดสินใจขั้นตอนถัดไปจนได้คำตอบ
เนื้อหาเกี่ยวข้อง — บทความที่เกี่ยวข้อง: SQLite Litestream Observability Stack — ระบบ
Automation Scripts ใช้ Agent ทำงานซ้ำๆอัตโนมัติ เช่น Data Collection, Report Generation, System Monitoring, Customer Support
เนื้อหาเกี่ยวข้อง — Ollama Local LLM Citizen Developer
LangChain Agent Setup
# === LangChain Agent Setup ===
# pip install langchain langchain-openai langchain-community
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_react_agent
from langchain.tools import tool
from langchain import hub
from typing import Optional
import json
# 1. สร้าง Custom Tools
@tool
def search_database(query: str) -> str:
"""ค้นหาข้อมูลในฐานข้อมูล ใช้เมื่อต้องการข้อมูลสินค้า ลูกค้า หรือ Orders"""
# Simulated database search
data = {
"products": {"count": 1500, "categories": 12},
"customers": {"count": 25000, "active": 18000},
"orders": {"today": 150, "total": 500000},
}
for key, value in data.items():
if key in query.lower():
return json.dumps(value, indent=2)
return "ไม่พบข้อมูลที่ค้นหา"
@tool
def calculate(expression: str) -> str:
"""คำนวณ Math expression เช่น '2+2', '100*0.15', 'sqrt(144)'"""
try:
import math
allowed = {"sqrt": math.sqrt, "pow": pow, "abs": abs,
"round": round, "min": min, "max": max}
result = eval(expression, {"__builtins__": {}}, allowed)
return str(result)
except Exception as e:
return f"Error: {e}"
@tool
def get_system_status(service: str) -> str:
"""ตรวจสอบสถานะระบบ ใช้เมื่อต้องการรู้ว่า service ทำงานปกติหรือไม่"""
statuses = {
"api": {"status": "healthy", "uptime": "99.9%", "latency": "45ms"},
"database": {"status": "healthy", "connections": 42, "cpu": "35%"},
"cache": {"status": "healthy", "hit_rate": "94%", "memory": "2.1GB"},
"queue": {"status": "warning", "pending": 1500, "consumers": 3},
}
if service.lower() in statuses:
return json.dumps(statuses[service.lower()], indent=2)
return f"Service '{service}' not found. Available: {list(statuses.keys())}"
@tool
def send_notification(message: str) -> str:
"""ส่ง Notification ไปยังทีม ใช้เมื่อต้องแจ้งเตือนเรื่องสำคัญ"""
print(f" [Notification] {message}")
return f"Notification sent: {message}"
# 2. สร้าง Agent
# llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# tools = [search_database, calculate, get_system_status, send_notification]
# prompt = hub.pull("hwchase17/react")
# agent = create_react_agent(llm, tools, prompt)
# agent_executor = AgentExecutor(
# agent=agent, tools=tools,
# verbose=True, max_iterations=5,
# handle_parsing_errors=True,
# )
# 3. รัน Agent
# result = agent_executor.invoke({
# "input": "ตรวจสอบสถานะ queue ถ้ามี pending มากกว่า 1000 ให้แจ้งทีม"
# })
# print(result["output"])
print("LangChain Agent Setup:")
print(" Tools: search_database, calculate, get_system_status, send_notification")
print(" Agent: ReAct Pattern")
print(" LLM: GPT-4o-mini")
print(" Max Iterations: 5")Automation Scripts
# automation_scripts.py — LangChain Agent Automation
from dataclasses import dataclass, field
from typing import List, Dict, Callable
from datetime import datetime
import json
@dataclass
class AutomationTask:
name: str
description: str
schedule: str # cron expression
prompt_template: str
tools_required: List[str]
last_run: str = ""
status: str = "pending"
class AutomationEngine:
"""LangChain Agent Automation Engine"""
def __init__(self):
self.tasks: List[AutomationTask] = []
self.execution_log: List[Dict] = []
def register_task(self, task: AutomationTask):
self.tasks.append(task)
def execute_task(self, task: AutomationTask):
"""Execute Automation Task"""
print(f"\n Executing: {task.name}")
print(f" Description: {task.description}")
print(f" Tools: {', '.join(task.tools_required)}")
# Agent execution simulation
# result = agent_executor.invoke({"input": task.prompt_template})
task.last_run = datetime.now().isoformat()
task.status = "completed"
self.execution_log.append({
"task": task.name,
"timestamp": task.last_run,
"status": "success",
})
print(f" Status: {task.status}")
print(f" Time: {task.last_run}")
def run_all(self):
"""Run All Automation Tasks"""
print(f"\n{'='*55}")
print(f"Automation Engine — {datetime.now().strftime('%Y-%m-%d %H:%M')}")
print(f"{'='*55}")
print(f" Tasks: {len(self.tasks)}")
for task in self.tasks:
self.execute_task(task)
print(f"\n Summary: {len(self.execution_log)} tasks executed")
def dashboard(self):
"""Automation Dashboard"""
print(f"\n Registered Tasks:")
for task in self.tasks:
print(f" [{task.status:>9}] {task.name}")
print(f" Schedule: {task.schedule}")
if task.last_run:
print(f" Last Run: {task.last_run}")
# สร้าง Automation Tasks
engine = AutomationEngine()
tasks = [
AutomationTask(
"Daily System Health Check",
"ตรวจสอบสถานะทุก Service แจ้งเตือนถ้ามีปัญหา",
"0 8 * * *",
"ตรวจสอบสถานะ api, database, cache, queue ถ้ามี service ไหนไม่ healthy ให้แจ้งทีม",
["get_system_status", "send_notification"],
),
AutomationTask(
"Sales Report Generation",
"สร้าง Daily Sales Report จาก Database",
"0 9 * * *",
"ดึงข้อมูล orders วันนี้ คำนวณยอดรวม สร้าง Summary Report",
["search_database", "calculate"],
),
AutomationTask(
"Customer Inquiry Response",
"ตอบคำถามลูกค้าอัตโนมัติจาก Queue",
"*/15 * * * *",
"ตรวจสอบ queue ถ้ามี pending inquiries ให้ตอบตาม Knowledge Base",
["search_database", "get_system_status"],
),
AutomationTask(
"Data Quality Check",
"ตรวจสอบคุณภาพข้อมูลใน Database",
"0 6 * * *",
"ค้นหา products ที่ไม่มี description, ลูกค้าที่ email ไม่ถูกต้อง แจ้งทีม Data",
["search_database", "send_notification"],
),
]
for task in tasks:
engine.register_task(task)
engine.run_all()
engine.dashboard()Advanced Agent Patterns
# advanced_agents.py — Advanced LangChain Agent Patterns
from dataclasses import dataclass, field
from typing import List, Dict, Optional
from enum import Enum
class AgentType(Enum):
REACT = "ReAct"
PLAN_AND_EXECUTE = "Plan and Execute"
MULTI_AGENT = "Multi-Agent"
TOOL_CALLING = "Tool Calling"
@dataclass
class AgentConfig:
name: str
agent_type: AgentType
llm_model: str
tools: List[str]
memory_type: str = "buffer" # buffer, summary, vector
max_iterations: int = 10
temperature: float = 0
class AgentPatterns:
"""Advanced Agent Design Patterns"""
def __init__(self):
self.patterns: Dict[str, AgentConfig] = {}
def register(self, config: AgentConfig):
self.patterns[config.name] = config
def show_patterns(self):
"""แสดง Agent Patterns"""
print(f"\n{'='*55}")
print(f"LangChain Agent Patterns")
print(f"{'='*55}")
for name, config in self.patterns.items():
print(f"\n [{config.agent_type.value}] {name}")
print(f" LLM: {config.llm_model}")
print(f" Tools: {', '.join(config.tools)}")
print(f" Memory: {config.memory_type}")
print(f" Max Iterations: {config.max_iterations}")
# LangGraph Multi-Agent Example
# from langgraph.graph import StateGraph, MessagesState
#
# def researcher(state: MessagesState):
# """Research Agent — ค้นหาข้อมูล"""
# messages = state["messages"]
# # Use search tools
# return {"messages": messages + [research_result]}
#
# def analyzer(state: MessagesState):
# """Analyzer Agent — วิเคราะห์ข้อมูล"""
# messages = state["messages"]
# # Analyze data
# return {"messages": messages + [analysis_result]}
#
# def writer(state: MessagesState):
# """Writer Agent — เขียน Report"""
# messages = state["messages"]
# # Write report
# return {"messages": messages + [report]}
#
# # สร้าง Graph
# graph = StateGraph(MessagesState)
# graph.add_node("researcher", researcher)
# graph.add_node("analyzer", analyzer)
# graph.add_node("writer", writer)
# graph.add_edge("researcher", "analyzer")
# graph.add_edge("analyzer", "writer")
# graph.set_entry_point("researcher")
# graph.set_finish_point("writer")
# app = graph.compile()
patterns = AgentPatterns()
configs = [
AgentConfig("Simple Q&A Agent", AgentType.REACT,
"gpt-4o-mini", ["search", "calculator"], "buffer", 5),
AgentConfig("Research Assistant", AgentType.PLAN_AND_EXECUTE,
"gpt-4o", ["search", "web_scraper", "summarizer"], "summary", 15),
AgentConfig("DevOps Automation", AgentType.TOOL_CALLING,
"gpt-4o-mini", ["system_status", "deploy", "notification"], "buffer", 10),
AgentConfig("Multi-Agent Pipeline", AgentType.MULTI_AGENT,
"gpt-4o", ["researcher", "analyzer", "writer"], "vector", 20),
]
for config in configs:
patterns.register(config)
patterns.show_patterns()
# Memory Types
memory_comparison = {
"ConversationBufferMemory": "เก็บทุก Message ง่าย ใช้ Token มาก",
"ConversationSummaryMemory": "สรุป Conversation ประหยัด Token",
"ConversationBufferWindowMemory": "เก็บ N Messages ล่าสุด",
"VectorStoreRetrieverMemory": "ค้นหา Relevant Memory ด้วย Similarity",
}
print(f"\n Memory Types:")
for mem_type, desc in memory_comparison.items():
print(f" {mem_type}: {desc}")Best Practices
- Tool Description: เขียน Description ให้ชัดเจน Agent จะเลือก Tool ตาม Description
- Max Iterations: ตั้ง max_iterations ป้องกัน Agent วนไม่รู้จบ
- Error Handling: ใช้ handle_parsing_errors=True จัดการ Output ผิดรูปแบบ
- Temperature 0: ใช้ temperature=0 สำหรับ Agent ให้ผลลัพธ์สม่ำเสมอ
- Structured Output: ใช้ Tool Calling Agent แทน ReAct สำหรับ Production
- Logging: Log ทุก Step ของ Agent สำหรับ Debugging
LangChain Agent คืออะไร
AI ใช้ LLM ตัดสินใจเลือก Tools Actions อัตโนมัติ วิเคราะห์คำถาม เลือก Tool เรียกใช้ ดูผลลัพธ์ ตัดสินใจขั้นตอนถัดไปจนได้คำตอบสุดท้าย
แนะนำเพิ่มเติม — บทวิเคราะห์จาก XM Signal
เนื้อหาเกี่ยวข้อง — แนะนำให้อ่าน price structure — ข้อมูลครบถ้วน 2026
