LangChain Agent
LangChain Agent ใช้ LLM ตัดสินใจเลือก Tools และ Actions อัตโนมัติ วิเคราะห์คำถาม เลือก Tool เรียกใช้ ดูผลลัพธ์ ตัดสินใจขั้นตอนถัดไปจนได้คำตอบ
Automation Scripts ใช้ Agent ทำงานซ้ำๆอัตโนมัติ เช่น Data Collection, Report Generation, System Monitoring, Customer Support
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 เรียกใช้ ดูผลลัพธ์ ตัดสินใจขั้นตอนถัดไปจนได้คำตอบสุดท้าย
LangChain Tools คืออะไร
Functions ที่ Agent เรียกใช้ Search Calculator Python REPL SQL Web Scraping API Calls Custom Tools @tool decorator กำหนด name description args
ReAct Agent ทำงานอย่างไร
Reasoning Acting 3 ขั้นตอนวนซ้ำ Thought วิเคราะห์ Action เลือกเรียก Tool Observation ดูผลลัพธ์ วนกลับ Thought จนได้ Final Answer Pattern นิยมที่สุด
วิธีสร้าง Custom Tool ทำอย่างไร
@tool decorator function รับ input string return string docstring อธิบาย Tool Agent อ่าน description ตัดสินใจใช้เมื่อไร API calls Database queries File operations
สรุป
LangChain Agent ใช้ LLM ตัดสินใจเลือก Tools อัตโนมัติ ReAct Pattern Thought Action Observation Custom Tools ด้วย @tool decorator Automation Scripts ทำงานซ้ำอัตโนมัติ Advanced Patterns Multi-Agent LangGraph Memory Types ตาม Use Case
