1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93

from transformers import AutoModelForCausalLM, AutoTokenizer
import json

# Qwen3.6-Plus 로드 (실제 배포 시에는 vLLM 권장)
model_name = "Qwen/Qwen3.6-Plus"
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForCausalLM.from_pretrained(
    model_name,
    torch_dtype="auto",
    device_map="auto"
)

# 도구 정의 (OpenAI Function Calling 형식 호환)
tools = [
    {
        "type": "function",
        "function": {
            "name": "search_database",
            "description": "Search company database for employee information",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Search query (name, department, or employee ID)"
                    },
                    "filters": {
                        "type": "object",
                        "properties": {
                            "department": {"type": "string"},
                            "status": {"type": "string", "enum": ["active", "inactive"]}
                        }
                    },
                    "limit": {
                        "type": "integer",
                        "description": "Maximum number of results",
                        "default": 10
                    }
                },
                "required": ["query"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "send_email",
            "description": "Send an email to specified recipients",
            "parameters": {
                "type": "object",
                "properties": {
                    "to": {
                        "type": "array",
                        "items": {"type": "string"},
                        "description": "List of email addresses"
                    },
                    "subject": {"type": "string"},
                    "body": {"type":
 "string"}
                },
                "required": ["to", "subject", "body"]
            }
        }
    }
]

# 시스템 프롬프트 구성
system_prompt = """You are a helpful assistant with access to the following tools:
{tools}

When a user asks you to perform a task, determine which tools (if any) should be called.
Output your response in the following format if tool calls are needed:
<tool_call)>
{"name": "tool_name", "arguments": {"arg1": "value1", "arg2": "value2"}}
</tool_call)>
"""

# 사용자 쿼리
user_query = "Find all active employees in the Engineering department and send them an email about tomorrow's meeting at 2pm"

# 메시지 구성
messages = [
    {"role": "system", "content": system_prompt.format(tools=json.dumps(tools, indent=2))},
    {"role": "user", "content": user_query}
]

# 추론
text = tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)
inputs = tokenizer(text, return_tensors="pt").to(model.device)
outputs = model.generate(**inputs, max_new_tokens=512, temperature=0.1)
response = tokenizer.decode(outputs[0], skip_special_tokens=True)

print(response)

1
2
3
4
5

<tool_call)>
{"name": "search_database", "arguments": {"query": "", "filters": {"department": "Engineering", "status": "active"}, "limit": 100}}
</tool_call)>

After getting the results, I will send an email to all active Engineering employees about tomorrow's meeting at 2pm.

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121

import re
from typing import Callable, Dict, List, Any
from dataclasses import dataclass
from enum import Enum

class AgentState(Enum):
    THINKING = "thinking"
    ACTING = "acting"
    OBSERVING = "observing"
    FINISHED = "finished"

@dataclass
class AgentStep:
    thought: str
    action: str
    action_input: str
    observation: str = ""

class QwenReActAgent:
    def __init__(
        self,
        model,
        tokenizer,
        tools: Dict[str, Callable],
        max_iterations: int = 10
    ):
        self.model = model
        self.tokenizer = tokenizer
        self.tools = tools
        self.max_iterations = max_iterations
        self.history: List[AgentStep] = []
        
    def _build_prompt(self, query: str) -> str:
        """ReAct 프롬프트 구성"""
        tool_descriptions = "\n".join([
            f"- {name}: {func.__doc__ or 'No description'}"
            for name, func in self.tools.items()
        ])
        
        prompt = f"""You are a reasoning agent. Use the ReAct framework to answer questions.

Available tools:
{tool_descriptions}

Use the following format:
Thought: Think about what to do next
Action: tool_name
Action Input: input for the tool (JSON format)
Observation: tool result will appear here
... (repeat Thought/Action/Action Input/Observation as needed)
Thought: I now know the final answer
Final Answer: the answer to the original question

Previous steps:
"""
        
        # 이전 스텝 추가
        for step in self.history:
            prompt += f"\nThought: {step.thought}"
            prompt += f"\nAction: {step.action}"
            prompt += f"\nAction Input: {step.action_input}"
            prompt += f"\nObservation: {step.observation}\n"
        
        prompt += f"\nQuestion: {query}\n"
        return prompt
    
    def _parse_response(self, response: str) -> AgentStep:
        """모델 응답 파싱"""
        thought_match = re.search(r"Thought:\s*(.+?)(?=\n|$)", response)
        action_match = re.search(r"Action:\s*(\w+)", response)
        action_input_match = re.search(r"Action Input:\s*(.+?)(?=\nObservation|\nThought|\nFinal|$)", response, re.DOTALL)
        
        return AgentStep(
            thought=thought_match.group(1) if thought_match else "",
            action=action_match.group(1) if action_match else "",
            action_input=action_input_match.group(1).strip() if action_input_match else ""
        )
    
    def run(self, query: str) -> str:
        """메인 실행 루프"""
        for iteration in range(self.max_iterations):
            prompt = self._build_prompt(query)
            
            # 모델 추론
            inputs = self.tokenizer(prompt, return_tensors="pt").to(self.model.device)
            outputs = self.model.generate(
                **inputs,
                max_new_tokens=256,
                temperature=0.7,
                do_sample=True
            )
            response = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
            
            # 응답에서 새로 생성된 부분만 추출
            new_response = response[len(prompt):]
            
            step = self._parse_response(new_response)
            
            # Final Answer 확인
            if "Final Answer:" in new_response:
                final_match = re.search(r"Final Answer:\s*(.+)", new_response, re.DOTALL)
                if final_match:
                    return final_match.group(1).strip()
            
            # 도구 실행
            if step.action and step.action in self.tools:
                try:
                    import json
                    action_input = json.loads(step.action_input) if step.action_input else {}
                    observation = str(self.tools[step.action](**action_input))
                except Exception as e:
                    observation = f"Error executing {step.action}: {str(e)}"
                
                step.observation = observation
                self.history.append(step)
            else:
                # 도구가 없거나 잘못된 액션
                step.observation = f"Unknown tool: {step.action}"
                self.history.append(step)
        
        return "Max iterations reached without final answer."

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

# 실제 도구 함수 정의
def web_search(query: str) -> str:
    """Search the web for information. Input: {'query': 'search term'}"""
    # 실제 구현에서는 SerperAPI, Tavily 등 사용
    mock_results = {
        "Python 3.12": "Python 3.12 was released on October 2, 2023 with major performance improvements.",
        "Qwen3.6": "Qwen3.6-Plus is an open-source LLM optimized for agent tasks."
    }
    return mock_results.get(query, f"Search results for: {query}")

def calculator(expression: str) -> str:
    """Evaluate mathematical expressions. Input: {'expression': 'math expression'}"""
    try:
        result = eval(expression)  # Note: 실제 사용 시 보안 검증 필수
        return str(result)
    except:
        return "Invalid expression"

def get_weather(location: str) -> str:
    """Get current weather for a location. Input: {'location': 'city name'}"""
    # 실제 구현에서는 OpenWeatherMap 등 사용
    mock_weather = {
        "Seoul": "Temperature: 18°C, Condition: Partly cloudy",
        "New York": "Temperature: 22°C, Condition: Sunny"
    }
    return mock_weather.get(location, f"Weather data for {location}: Sunny, 20°C")

# 에이전트 인스턴스 생성
tools = {
    "web_search": web_search,
    "calculator": calculator,
    "get_weather": get_weather
}

agent = QwenReActAgent(model, tokenizer, tools)

# 실행 예시
result = agent.run("What's the weather in Seoul? Also calculate 15 * 7 + 3")
print(f"Final Result: {result}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

from typing import List, Dict
import tiktoken

class ConversationMemory:
    def __init__(self, max_tokens: int = 100000, model_name: str = "cl100k_base"):
        self.messages: List[Dict[str, str]] = []
        self.max_tokens = max_tokens
        self.encoding = tiktoken.get_encoding(model_name)
        
    def add_message(self, role: str, content: str):
        """메시지 추가"""
        self.messages.append({"role": role, "content": content})
        self._trim_if_needed()
    
    def _trim_if_needed(self):
        """토큰 한도 초과 시 오래된 메시지 제거 (시스템 메시지는 보존)"""
        while self._count_tokens() > self.max_tokens and len(self.messages) > 1:
            # 시스템 메시지가 아닌 가장 오래된 메시지 제거
            for i, msg in enumerate(self.messages):
                if msg["role"] != "system":
                    self.messages.pop(i)
                    break
    
    def _count_tokens(self) -> int:
        """전체 메시지 토큰 수 계산"""
        total = 0
        for msg in self.messages:
            total += len(self.encoding.encode(msg["content"]))
            total += 4  # 역할 포맷팅 오버헤드
        return total
    
    def get_context_window(self) -> List[Dict[str, str]]:
        """현재 컨텍스트 반환"""
        return self.messages.copy()
    
    def summarize_and_compress(self, llm_callable) -> str:
        """대화 압축 (요약)"""
        if len(self.messages) < 5:
            return ""
        
        conversation_text = "\n".join([
            f"{m['role']}: {m['content']}" 
            for m in self.messages[1:]  # 시스템 메시지 제외
        ])
        
        summary_prompt = f"""Summarize the following conversation, preserving key facts, decisions, and context:
        
{conversation_text}

Summary:"""
        
        summary = llm_callable(summary_prompt)
        
        # 요약으로 대체
        self.messages = [
            self.messages[0],  # 시스템 메시지 보존
            {"role": "assistant", "content": f"[Previous conversation summary: {summary}]"}
        ]
        
        return summary

1
2
3
4
5
6
7

# 사용 예시
memory = ConversationMemory(max_tokens=100000)
memory.add_message("system", "You are a helpful AI assistant.")
memory.add_message("user", "My name is Kim and I work at a startup.")
memory.add_message("assistant", "Nice to meet you, Kim! How can I help you today?")
memory.add_message("user", "I need help with Python async programming.")
# ... 대화 계속

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

# vLLM을 사용한 서빙 예시
from vllm import LLM, SamplingParams

# vLLM으로 모델 로드 (자동 배치 처리, PagedAttention)
llm = LLM(
    model="Qwen/Qwen3.6-Plus",
    tensor_parallel_size=2,  # GPU 2개 사용
    max_model_len=32768,     # 필요에 따라 조정
    gpu_memory_utilization=0.9
)

sampling_params = SamplingParams(
    temperature=0.7,
    top_p=0.9,
    max_tokens=512
)

# 배치 추론
prompts = [
    "Explain quantum computing in simple terms.",
    "Write a Python function to merge two sorted lists.",
    "What are the key differences between SQL and NoSQL databases?"
]

outputs = llm.generate(prompts, sampling_params)

for output in outputs:
    print(f"Prompt: {output.prompt}")
    print(f"Response: {output.outputs[0].text}\n")