1"""
215. LLM ์์ด์ ํธ (LLM Agents) ์์
3
4ReAct ํจํด, Tool Use, LangChain Agent ์ค์ต
5"""
6
7import json
8import re
9from typing import Dict, List, Callable, Any
10
11print("=" * 60)
12print("LLM ์์ด์ ํธ (LLM Agents)")
13print("=" * 60)
14
15
16# ============================================
17# 1. ๋๊ตฌ ์ ์
18# ============================================
19print("\n[1] ๋๊ตฌ (Tools) ์ ์")
20print("-" * 40)
21
22class Tool:
23 """๋๊ตฌ ๊ธฐ๋ณธ ํด๋์ค"""
24
25 def __init__(self, name: str, description: str, func: Callable):
26 self.name = name
27 self.description = description
28 self.func = func
29
30 def run(self, input_str: str) -> str:
31 """๋๊ตฌ ์คํ"""
32 try:
33 return str(self.func(input_str))
34 except Exception as e:
35 return f"Error: {str(e)}"
36
37
38# ๋๊ตฌ ๊ตฌํ
39def calculator(expression: str) -> float:
40 """์์ ํ ์ํ ๊ณ์ฐ"""
41 # ๊ฐ๋จํ ์์ ์ฒดํฌ
42 allowed_chars = set("0123456789+-*/.() ")
43 if not all(c in allowed_chars for c in expression):
44 raise ValueError("Invalid characters in expression")
45 return eval(expression)
46
47def search(query: str) -> str:
48 """๊ฒ์ ์๋ฎฌ๋ ์ด์
"""
49 # ์ค์ ๋ก๋ API ํธ์ถ
50 results = {
51 "ํ์ด์ฌ ์ฐฝ์์": "ํ์ด์ฌ์ 1991๋
๊ท๋ ๋ฐ ๋ก์ฌ(Guido van Rossum)์ด ๊ฐ๋ฐํ์ต๋๋ค.",
52 "์ธ๊ณต์ง๋ฅ": "์ธ๊ณต์ง๋ฅ(AI)์ ์ธ๊ฐ์ ์ง๋ฅ์ ๋ชจ๋ฐฉํ๋ ์ปดํจํฐ ์์คํ
์
๋๋ค.",
53 "์์ธ ์ธ๊ตฌ": "์์ธ์ ์ธ๊ตฌ๋ ์ฝ 950๋ง ๋ช
์
๋๋ค (2024๋
๊ธฐ์ค).",
54 }
55 for key, value in results.items():
56 if key in query:
57 return value
58 return f"'{query}'์ ๋ํ ๊ฒ์ ๊ฒฐ๊ณผ๋ฅผ ์ฐพ์ ์ ์์ต๋๋ค."
59
60def get_weather(city: str) -> str:
61 """๋ ์จ ์กฐํ ์๋ฎฌ๋ ์ด์
"""
62 weather_data = {
63 "์์ธ": {"temp": 25, "condition": "๋ง์"},
64 "๋ถ์ฐ": {"temp": 28, "condition": "ํ๋ฆผ"},
65 "์ ์ฃผ": {"temp": 27, "condition": "๊ตฌ๋ฆ ์กฐ๊ธ"},
66 }
67 if city in weather_data:
68 data = weather_data[city]
69 return f"{city} ๋ ์จ: {data['temp']}๋, {data['condition']}"
70 return f"{city}์ ๋ ์จ ์ ๋ณด๋ฅผ ์ฐพ์ ์ ์์ต๋๋ค."
71
72
73# ๋๊ตฌ ๋ฑ๋ก
74tools = [
75 Tool("calculator", "์ํ ๊ณ์ฐ. ์
๋ ฅ: ์ํ ํํ์ (์: '2 + 3 * 4')", calculator),
76 Tool("search", "์ ๋ณด ๊ฒ์. ์
๋ ฅ: ๊ฒ์์ด", search),
77 Tool("get_weather", "๋ ์จ ์กฐํ. ์
๋ ฅ: ๋์ ์ด๋ฆ", get_weather),
78]
79
80print("์ฌ์ฉ ๊ฐ๋ฅํ ๋๊ตฌ:")
81for tool in tools:
82 print(f" - {tool.name}: {tool.description}")
83
84
85# ============================================
86# 2. ReAct ํจํด ์๋ฎฌ๋ ์ด์
87# ============================================
88print("\n[2] ReAct ํจํด ์๋ฎฌ๋ ์ด์
")
89print("-" * 40)
90
91class ReActAgent:
92 """ReAct (Reasoning + Acting) ์์ด์ ํธ"""
93
94 def __init__(self, tools: List[Tool]):
95 self.tools = {t.name: t for t in tools}
96 self.history = []
97
98 def think(self, question: str, observations: List[str]) -> Dict[str, str]:
99 """
100 ์๊ฐ ๋จ๊ณ (์ค์ ๋ก๋ LLM ํธ์ถ)
101 ์ฌ๊ธฐ์๋ ๊ท์น ๊ธฐ๋ฐ์ผ๋ก ์๋ฎฌ๋ ์ด์
102 """
103 question_lower = question.lower()
104
105 # ๊ท์น ๊ธฐ๋ฐ ์์ฌ๊ฒฐ์ (์ค์ ๋ก๋ LLM์ด ์ํ)
106 if "๋ ์จ" in question_lower:
107 # ๋์ ์ถ์ถ
108 cities = ["์์ธ", "๋ถ์ฐ", "์ ์ฃผ"]
109 for city in cities:
110 if city in question:
111 return {
112 "thought": f"{city}์ ๋ ์จ๋ฅผ ํ์ธํด์ผ ํฉ๋๋ค.",
113 "action": "get_weather",
114 "action_input": city
115 }
116
117 if "๊ณ์ฐ" in question_lower or any(op in question for op in ["+", "-", "*", "/"]):
118 # ์์ ์ถ์ถ
119 numbers = re.findall(r'[\d\+\-\*\/\.\(\)\s]+', question)
120 if numbers:
121 expr = numbers[0].strip()
122 return {
123 "thought": f"'{expr}'์ ๊ณ์ฐํด์ผ ํฉ๋๋ค.",
124 "action": "calculator",
125 "action_input": expr
126 }
127
128 if any(keyword in question_lower for keyword in ["๋๊ตฌ", "๋ฌด์", "์ด๋", "๊ฒ์"]):
129 return {
130 "thought": f"'{question}'์ ๋ํด ๊ฒ์ํด์ผ ํฉ๋๋ค.",
131 "action": "search",
132 "action_input": question
133 }
134
135 # ์ด๋ฏธ ์ถฉ๋ถํ ์ ๋ณด๊ฐ ์์ผ๋ฉด ์ต์ข
๋ต๋ณ
136 if observations:
137 return {
138 "thought": "์ถฉ๋ถํ ์ ๋ณด๋ฅผ ์์งํ์ต๋๋ค.",
139 "final_answer": " ".join(observations)
140 }
141
142 return {
143 "thought": "์ง๋ฌธ์ ์ดํดํ์ง ๋ชปํ์ต๋๋ค.",
144 "final_answer": "์ฃ์กํฉ๋๋ค, ์ง๋ฌธ์ ์ฒ๋ฆฌํ ์ ์์ต๋๋ค."
145 }
146
147 def act(self, action: str, action_input: str) -> str:
148 """ํ๋ ๋จ๊ณ"""
149 if action in self.tools:
150 return self.tools[action].run(action_input)
151 return f"Error: Unknown tool '{action}'"
152
153 def run(self, question: str, max_steps: int = 5) -> str:
154 """์์ด์ ํธ ์คํ"""
155 observations = []
156
157 print(f"\n์ง๋ฌธ: {question}")
158 print("-" * 30)
159
160 for step in range(max_steps):
161 # ์๊ฐ
162 result = self.think(question, observations)
163
164 print(f"\n[Step {step + 1}]")
165 print(f"Thought: {result.get('thought', '')}")
166
167 # ์ต์ข
๋ต๋ณ ํ์ธ
168 if "final_answer" in result:
169 print(f"Final Answer: {result['final_answer']}")
170 return result["final_answer"]
171
172 # ํ๋
173 action = result.get("action")
174 action_input = result.get("action_input")
175
176 if action:
177 print(f"Action: {action}")
178 print(f"Action Input: {action_input}")
179
180 observation = self.act(action, action_input)
181 observations.append(observation)
182 print(f"Observation: {observation}")
183
184 return "์ต๋ ๋จ๊ณ ๋๋ฌ"
185
186
187# ํ
์คํธ
188agent = ReActAgent(tools)
189
190questions = [
191 "์์ธ ๋ ์จ ์ด๋?",
192 "15 + 27 * 3์ ๊ณ์ฐํด์ค",
193 "ํ์ด์ฌ ์ฐฝ์์๊ฐ ๋๊ตฌ์ผ?",
194]
195
196for q in questions:
197 result = agent.run(q)
198
199
200# ============================================
201# 3. Function Calling ํ์
202# ============================================
203print("\n" + "=" * 60)
204print("[3] Function Calling ํ์")
205print("-" * 40)
206
207# OpenAI Function Calling ํ์
208function_definitions = [
209 {
210 "type": "function",
211 "function": {
212 "name": "get_weather",
213 "description": "ํน์ ๋์์ ํ์ฌ ๋ ์จ ์ ๋ณด๋ฅผ ๊ฐ์ ธ์ต๋๋ค.",
214 "parameters": {
215 "type": "object",
216 "properties": {
217 "city": {
218 "type": "string",
219 "description": "๋์ ์ด๋ฆ (์: Seoul, Tokyo)"
220 },
221 "unit": {
222 "type": "string",
223 "enum": ["celsius", "fahrenheit"],
224 "description": "์จ๋ ๋จ์"
225 }
226 },
227 "required": ["city"]
228 }
229 }
230 },
231 {
232 "type": "function",
233 "function": {
234 "name": "search_web",
235 "description": "์น์์ ์ ๋ณด๋ฅผ ๊ฒ์ํฉ๋๋ค.",
236 "parameters": {
237 "type": "object",
238 "properties": {
239 "query": {
240 "type": "string",
241 "description": "๊ฒ์์ด"
242 }
243 },
244 "required": ["query"]
245 }
246 }
247 }
248]
249
250print("Function Calling ์ ์:")
251print(json.dumps(function_definitions, indent=2, ensure_ascii=False))
252
253
254# ============================================
255# 4. ๋ฉํฐ ์์ด์ ํธ ์๋ฎฌ๋ ์ด์
256# ============================================
257print("\n" + "=" * 60)
258print("[4] ๋ฉํฐ ์์ด์ ํธ ์๋ฎฌ๋ ์ด์
")
259print("-" * 40)
260
261class ResearcherAgent:
262 """์ฐ๊ตฌ ์์ด์ ํธ"""
263
264 def __init__(self):
265 self.name = "Researcher"
266
267 def research(self, topic: str) -> str:
268 """์ฃผ์ ์ฐ๊ตฌ (์๋ฎฌ๋ ์ด์
)"""
269 research_db = {
270 "์ธ๊ณต์ง๋ฅ": "AI๋ 1956๋
๋คํธ๋จธ์ค ํ์์์ ์์๋์์ผ๋ฉฐ, "
271 "๋จธ์ ๋ฌ๋, ๋ฅ๋ฌ๋, ์์ฐ์ด์ฒ๋ฆฌ ๋ฑ์ผ๋ก ๋ฐ์ ํ์ต๋๋ค.",
272 "ํ์ด์ฌ": "ํ์ด์ฌ์ 1991๋
๊ท๋ ๋ฐ ๋ก์ฌ์ด ๊ฐ๋ฐํ ํ๋ก๊ทธ๋๋ฐ ์ธ์ด๋ก, "
273 "๊ฐ๊ฒฐํ ๋ฌธ๋ฒ๊ณผ ํ๋ถํ ๋ผ์ด๋ธ๋ฌ๋ฆฌ๊ฐ ํน์ง์
๋๋ค.",
274 }
275 for key, value in research_db.items():
276 if key in topic:
277 return value
278 return f"{topic}์ ๋ํ ์ฐ๊ตฌ ๊ฒฐ๊ณผ๋ฅผ ์ฐพ์ ์ ์์ต๋๋ค."
279
280
281class WriterAgent:
282 """์๋ฌธ ์์ด์ ํธ"""
283
284 def __init__(self):
285 self.name = "Writer"
286
287 def write(self, research_results: str, style: str = "formal") -> str:
288 """๋ฌธ์ ์์ฑ (์๋ฎฌ๋ ์ด์
)"""
289 if style == "formal":
290 return f"## ์ฐ๊ตฌ ๋ณด๊ณ ์\n\n{research_results}\n\n๋ณธ ๋ด์ฉ์ ์ฐ๊ตฌ ๊ฒฐ๊ณผ๋ฅผ ๋ฐํ์ผ๋ก ์์ฑ๋์์ต๋๋ค."
291 else:
292 return f"# ์์ฝ\n\n{research_results}"
293
294
295class ReviewerAgent:
296 """๊ฒํ ์์ด์ ํธ"""
297
298 def __init__(self):
299 self.name = "Reviewer"
300
301 def review(self, document: str) -> str:
302 """๋ฌธ์ ๊ฒํ (์๋ฎฌ๋ ์ด์
)"""
303 issues = []
304 if len(document) < 100:
305 issues.append("๋ด์ฉ์ด ๋๋ฌด ์งง์ต๋๋ค.")
306 if "์ฐธ๊ณ ๋ฌธํ" not in document:
307 issues.append("์ฐธ๊ณ ๋ฌธํ์ด ์์ต๋๋ค.")
308
309 if issues:
310 return "๊ฒํ ๊ฒฐ๊ณผ:\n" + "\n".join(f"- {issue}" for issue in issues)
311 return "๊ฒํ ๊ฒฐ๊ณผ: ์์ ํ์ ์์"
312
313
314class MultiAgentSystem:
315 """๋ฉํฐ ์์ด์ ํธ ์์คํ
"""
316
317 def __init__(self):
318 self.researcher = ResearcherAgent()
319 self.writer = WriterAgent()
320 self.reviewer = ReviewerAgent()
321
322 def create_document(self, topic: str) -> str:
323 """๋ฌธ์ ์์ฑ ํ์ดํ๋ผ์ธ"""
324 print(f"\nํ ํฝ: {topic}")
325
326 # 1. ์ฐ๊ตฌ
327 print(f"\n[{self.researcher.name}] ์ฐ๊ตฌ ์ค...")
328 research = self.researcher.research(topic)
329 print(f"์ฐ๊ตฌ ๊ฒฐ๊ณผ: {research[:50]}...")
330
331 # 2. ์์ฑ
332 print(f"\n[{self.writer.name}] ์์ฑ ์ค...")
333 document = self.writer.write(research)
334 print(f"์์ฑ ๊ฒฐ๊ณผ: {document[:50]}...")
335
336 # 3. ๊ฒํ
337 print(f"\n[{self.reviewer.name}] ๊ฒํ ์ค...")
338 review = self.reviewer.review(document)
339 print(f"๊ฒํ ๊ฒฐ๊ณผ: {review}")
340
341 return document
342
343
344# ํ
์คํธ
345system = MultiAgentSystem()
346doc = system.create_document("์ธ๊ณต์ง๋ฅ์ ์ญ์ฌ")
347print(f"\n์ต์ข
๋ฌธ์:\n{doc}")
348
349
350# ============================================
351# 5. LangChain Agent ์ฝ๋ ์์
352# ============================================
353print("\n" + "=" * 60)
354print("[5] LangChain Agent ์ฝ๋ ์์")
355print("-" * 40)
356
357langchain_code = '''
358from langchain_openai import ChatOpenAI
359from langchain.agents import AgentExecutor, create_openai_tools_agent
360from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
361from langchain.tools import tool
362
363# LLM
364llm = ChatOpenAI(model="gpt-4", temperature=0)
365
366# ๋๊ตฌ ์ ์
367@tool
368def calculator(expression: str) -> str:
369 """์ํ ๊ณ์ฐ. ์
๋ ฅ: ์ํ ํํ์ (์: '2 + 3 * 4')"""
370 return str(eval(expression))
371
372@tool
373def get_current_time() -> str:
374 """ํ์ฌ ์๊ฐ ๋ฐํ"""
375 from datetime import datetime
376 return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
377
378tools = [calculator, get_current_time]
379
380# ํ๋กฌํํธ
381prompt = ChatPromptTemplate.from_messages([
382 ("system", "๋น์ ์ ๋์์ด ๋๋ AI์
๋๋ค."),
383 MessagesPlaceholder(variable_name="chat_history", optional=True),
384 ("human", "{input}"),
385 MessagesPlaceholder(variable_name="agent_scratchpad"),
386])
387
388# ์์ด์ ํธ ์์ฑ
389agent = create_openai_tools_agent(llm, tools, prompt)
390agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
391
392# ์คํ
393result = agent_executor.invoke({"input": "ํ์ฌ ์๊ฐ๊ณผ 15 + 27์ ๊ฒฐ๊ณผ๋ฅผ ์๋ ค์ค"})
394print(result["output"])
395'''
396print(langchain_code)
397
398
399# ============================================
400# 6. ์์จ ์์ด์ ํธ (AutoGPT ์คํ์ผ)
401# ============================================
402print("\n" + "=" * 60)
403print("[6] ์์จ ์์ด์ ํธ (AutoGPT ์คํ์ผ)")
404print("-" * 40)
405
406class AutoGPTLikeAgent:
407 """AutoGPT ์คํ์ผ ์์จ ์์ด์ ํธ"""
408
409 def __init__(self, tools: List[Tool], goals: List[str]):
410 self.tools = {t.name: t for t in tools}
411 self.goals = goals
412 self.memory = []
413 self.completed_tasks = []
414
415 def plan(self) -> Dict[str, Any]:
416 """๋ค์ ์์
๊ณํ (๊ท์น ๊ธฐ๋ฐ ์๋ฎฌ๋ ์ด์
)"""
417 # ์์ง ๋ฌ์ฑํ์ง ์์ ๋ชฉํ ํ์ธ
418 remaining_goals = [g for g in self.goals if g not in self.completed_tasks]
419
420 if not remaining_goals:
421 return {"task": "COMPLETE", "summary": "๋ชจ๋ ๋ชฉํ ๋ฌ์ฑ!"}
422
423 current_goal = remaining_goals[0]
424
425 # ๊ฐ๋จํ ๊ท์น ๊ธฐ๋ฐ ๊ณํ
426 if "๋ ์จ" in current_goal:
427 return {
428 "task": current_goal,
429 "tool": "get_weather",
430 "input": "์์ธ"
431 }
432 elif "๊ณ์ฐ" in current_goal or "์ซ์" in current_goal:
433 return {
434 "task": current_goal,
435 "tool": "calculator",
436 "input": "100 + 200"
437 }
438 else:
439 return {
440 "task": current_goal,
441 "tool": "search",
442 "input": current_goal
443 }
444
445 def execute(self, plan: Dict[str, Any]) -> Dict[str, Any]:
446 """๊ณํ ์คํ"""
447 if plan.get("task") == "COMPLETE":
448 return {"status": "complete", "summary": plan["summary"]}
449
450 tool_name = plan.get("tool")
451 tool_input = plan.get("input")
452
453 if tool_name in self.tools:
454 result = self.tools[tool_name].run(tool_input)
455 return {"status": "success", "result": result}
456
457 return {"status": "error", "message": f"Unknown tool: {tool_name}"}
458
459 def run(self, max_iterations: int = 5) -> str:
460 """์์ด์ ํธ ์คํ"""
461 print(f"\n๋ชฉํ: {self.goals}")
462 print("-" * 30)
463
464 for i in range(max_iterations):
465 print(f"\n=== Iteration {i + 1} ===")
466
467 # ๊ณํ
468 plan = self.plan()
469 print(f"Plan: {plan}")
470
471 if plan.get("task") == "COMPLETE":
472 print(f"์๋ฃ: {plan['summary']}")
473 return plan["summary"]
474
475 # ์คํ
476 result = self.execute(plan)
477 print(f"Result: {result}")
478
479 # ๋ฉ๋ชจ๋ฆฌ ๋ฐ ์๋ฃ ๋ชฉ๋ก ์
๋ฐ์ดํธ
480 self.memory.append({"plan": plan, "result": result})
481 if result["status"] == "success":
482 self.completed_tasks.append(plan["task"])
483
484 return "์ต๋ ๋ฐ๋ณต ํ์ ๋๋ฌ"
485
486
487# ํ
์คํธ
488auto_agent = AutoGPTLikeAgent(
489 tools=tools,
490 goals=["์์ธ ๋ ์จ ํ์ธ", "๊ฐ๋จํ ๊ณ์ฐ"]
491)
492auto_agent.run()
493
494
495# ============================================
496# ์ ๋ฆฌ
497# ============================================
498print("\n" + "=" * 60)
499print("LLM ์์ด์ ํธ ์ ๋ฆฌ")
500print("=" * 60)
501
502summary = """
503LLM ์์ด์ ํธ ํต์ฌ ๊ฐ๋
:
504
5051. ReAct ํจํด:
506 Thought -> Action -> Observation -> ... -> Final Answer
507
5082. ๋๊ตฌ ์ ์:
509 - ์ด๋ฆ: ๊ณ ์ ์๋ณ์
510 - ์ค๋ช
: LLM์ด ์ฌ์ฉ ์์ ํ๋จ์ฉ
511 - ํจ์: ์ค์ ์คํ ๋ก์ง
512
5133. Function Calling (OpenAI):
514 - tools ํ๋ผ๋ฏธํฐ๋ก ํจ์ ์ ์
515 - tool_choice="auto"๋ก ์๋ ์ ํ
516 - ๊ฒฐ๊ณผ๋ฅผ role="tool"๋ก ์ ๋ฌ
517
5184. ๋ฉํฐ ์์ด์ ํธ:
519 - ์ญํ ๋ถ๋ด (์ฐ๊ตฌ, ์์ฑ, ๊ฒํ )
520 - ํ์ดํ๋ผ์ธ ์ฐ๊ฒฐ
521 - ํ์
ํ๋กํ ์ฝ
522
5235. ์์จ ์์ด์ ํธ:
524 - ๋ชฉํ ๊ธฐ๋ฐ ๊ณํ
525 - ๋ฉ๋ชจ๋ฆฌ ์ ์ง
526 - ๋ฐ๋ณต์ ์คํ
527
528์์ด์ ํธ ์ค๊ณ ์ฒดํฌ๋ฆฌ์คํธ:
529โก ๋ช
ํํ ๋๊ตฌ ์ค๋ช
530โก ์๋ฌ ์ฒ๋ฆฌ
531โก ๋ฌดํ ๋ฃจํ ๋ฐฉ์ง (max_steps)
532โก ์์ ์ฅ์น (์ํํ ์์
์ ํ)
533โก ๋ก๊น
๋ฐ ๋๋ฒ๊น
534"""
535print(summary)