import json import requests from typing import Generator, Dict, Any # VLLM Configuration VLLM_BASE_URL = "http://localhost:8900" VLLM_API_KEY = "your-vllm-api-key-here" VLLM_MODEL = "openai/gpt-oss-20b" VLLM_TIMEOUT = 300 VLLM_MAX_RETRIES = 3 VLLM_CONTEXT_RATIO_LIMIT = 0.85 VLLM_TEMPERATURE = 0.8 VLLM_TOP_P = 0.8 VLLM_TOP_K = 40 VLLM_MIN_P = 0.05 VLLM_PRESENCE_PENALTY = 1.5 VLLM_FREQUENCY_PENALTY = 0.0 VLLM_REPETITION_PENALTY = 1.1 VLLM_MAX_TOKENS = 35000 VLLM_ENABLE_THINKING = True def stream_chat_completion(messages: list, tools: list = None) -> Generator[str, None, None]: """Stream chat completion from vLLM server""" url = f"{VLLM_BASE_URL}/v1/chat/completions" payload = { "model": VLLM_MODEL, "messages": messages, "stream": True, "temperature": VLLM_TEMPERATURE, "top_p": VLLM_TOP_P, "top_k": VLLM_TOP_K, "min_p": VLLM_MIN_P, "presence_penalty": VLLM_PRESENCE_PENALTY, "frequency_penalty": VLLM_FREQUENCY_PENALTY, "max_tokens": VLLM_MAX_TOKENS, "repetition_penalty": VLLM_REPETITION_PENALTY, "enable_thinking": VLLM_ENABLE_THINKING } if tools: payload["tools"] = tools payload["tool_choice"] = "auto" payload["reasoning_effort"] = "high" headers = { "Content-Type": "application/json", "Authorization": f"Bearer {VLLM_API_KEY}" } response = requests.post( url, json=payload, headers=headers, stream=True, timeout=VLLM_TIMEOUT ) response.raise_for_status() accumulated_tool_calls = {} for line in response.iter_lines(): if line: line_str = line.decode('utf-8') if line_str.startswith("data: "): data_str = line_str[6:] if data_str == "[DONE]": if accumulated_tool_calls: print("\n\n==== tool_calls ====") yield json.dumps(list(accumulated_tool_calls.values())) else: print("\n\n==== tool_calls not found ====\n\n") break try: data = json.loads(data_str) if "choices" in data and len(data["choices"]) > 0: choice = data["choices"][0] if "delta" in choice: if "content" in choice["delta"] and choice["delta"]["content"]: yield choice["delta"]["content"] if "reasoning_content" in choice["delta"] and choice["delta"]["reasoning_content"]: yield choice["delta"]["reasoning_content"] if "tool_calls" in choice["delta"]: for tool_call_delta in choice["delta"]["tool_calls"]: index = tool_call_delta.get("index", 0) if index not in accumulated_tool_calls: accumulated_tool_calls[index] = { "id": tool_call_delta.get("id", f"call_{index}"), "type": "function", "function": { "name": "", "arguments": "" } } if "function" in tool_call_delta: if "name" in tool_call_delta["function"]: accumulated_tool_calls[index]["function"]["name"] += tool_call_delta["function"]["name"] if "arguments" in tool_call_delta["function"]: accumulated_tool_calls[index]["function"]["arguments"] += tool_call_delta["function"]["arguments"] except json.JSONDecodeError: continue def execute_tool(tool_name: str, arguments: Dict[str, Any]) -> str: """Simulate tool execution""" if tool_name == "file_read": file_path = arguments.get("file_path", "unknown") start_line = arguments.get("start_line", 1) line_count = arguments.get("line_count", 100) return f"File read simulation: {file_path} from line {start_line}, count {line_count}" elif tool_name == "file_search": filename = arguments.get("filename", "unknown") return f"File search simulation: Found {filename} at /path/to/{filename}" elif tool_name == "project_tree": max_depth = arguments.get("max_depth", 10) return f"Project tree simulation with max depth {max_depth}" elif tool_name == "text_search": text = arguments.get("text", "unknown") return f"Text search simulation for: {text}" return f"Unknown tool: {tool_name}" def main(): # Load messages from messages.txt with open('messages.txt', 'r', encoding='utf-8') as f: messages = json.load(f) # Define available tools tools = [ { "type": "function", "function": { "name": "file_read", "description": "Reads the contents of a specified file (split into 100 lines by default). Must use absolute path found by file_search. Check line count from file_search results to optimize token usage - for large files read only necessary parts, for small files read entire content.", "parameters": { "type": "object", "properties": { "file_path": { "type": "string", "description": "Absolute path of the file to read (use file_search result, e.g., /home/user/project/src/main.java)" }, "start_line": { "type": "integer", "description": "Line number to start reading from (default: 1, starts from 1). Consider total line count confirmed in file_search when setting.", "default": 1 }, "line_count": { "type": "integer", "description": "Number of lines to read (default: 100, max: 500). If file_search result shows fewer lines read more, if many lines read only as needed to save tokens.", "default": 100 } }, "required": ["file_path"] } } }, { "type": "function", "function": { "name": "file_search", "description": "Find absolute path by filename. Must be used before file_read/file_edit. Search results include line count for each file.", "parameters": { "type": "object", "properties": { "filename": { "type": "string", "description": "Exact filename (e.g., Main.java, config.xml)" } }, "required": ["filename"] } } }, { "type": "function", "function": { "name": "project_tree", "description": "Output project structure in tree format. Used to understand structure.", "parameters": { "type": "object", "properties": { "max_depth": { "type": "integer", "description": "Exploration depth (default: 10)", "default": 10 }, "show_hidden": { "type": "boolean", "description": "Show hidden files (default: false)", "default": False }, "files_only": { "type": "boolean", "description": "Show files only (default: false)", "default": False }, "dirs_only": { "type": "boolean", "description": "Show directories only (default: false)", "default": False } }, "required": [] } } }, { "type": "function", "function": { "name": "text_search", "description": "Search text in project. Separate with commas for OR search.", "parameters": { "type": "object", "properties": { "text": { "type": "string", "description": "Search text. Single: getUserById, OR search: new Thread,@PostConstruct,print" } }, "required": ["text"] } } } ] print("=== response ===") for chunk in stream_chat_completion(messages, tools): print(chunk, end='', flush=True) if __name__ == "__main__": main()